@cryptotaxi247 / netdata-1 / commits / 8329088ca

fix(go/framweork): restore v2 host-scoped chart emission across host switches (#21979)

Ilya Mashchenko committed Mar 19, 2026 at 22:26 UTC 8329088caf5e01da2667b74951a9567f03afe89d
27 files changed +2199 -291
src/go/pkg/netdataapi/api.go
+10 -3
@@ -6,6 +6,7 @@ import (
6 "bytes"
7 "fmt"
8 "io"
9 + "sort"
10 "strconv"
11 )
12
@@ -126,9 +127,15 @@ func (a *API) EMPTYLINE() error {
127 func (a *API) HOSTINFO(info HostInfo) {
128 var buf bytes.Buffer
129
129 - buf.WriteString(fmt.Sprintf("HOST_DEFINE '%s' '%s'\n", info.GUID, info.Hostname))
130 - for k, v := range info.Labels {
131 - buf.WriteString(fmt.Sprintf("HOST_LABEL '%s' '%s'\n", k, v))
130 + _, _ = fmt.Fprintf(&buf, "HOST_DEFINE '%s' '%s'\n", info.GUID, info.Hostname)
131 + keys := make([]string, 0, len(info.Labels))
132 + for k := range info.Labels {
133 + keys = append(keys, k)
134 + }
135 + sort.Strings(keys)
136 + for _, k := range keys {
137 + v := info.Labels[k]
138 + _, _ = fmt.Fprintf(&buf, "HOST_LABEL '%s' '%s'\n", k, v)
139 }
140 buf.WriteString("HOST_DEFINE_END\n\n")
141
src/go/pkg/netdataapi/api_test.go
+2
@@ -210,6 +210,7 @@ func TestHOSTINFO(t *testing.T) {
210 GUID: "test-guid",
211 Hostname: "test-host",
212 Labels: map[string]string{
213 + "label2": "value2",
214 "label1": "value1",
215 },
216 }
@@ -219,6 +220,7 @@ func TestHOSTINFO(t *testing.T) {
220 expected := `
221 HOST_DEFINE 'test-guid' 'test-host'
222 HOST_LABEL 'label1' 'value1'
223 +HOST_LABEL 'label2' 'value2'
224 HOST_DEFINE_END
225
226 `[1:]
src/go/plugin/agent/runtimechartemit/components.go
+11 -16
@@ -4,6 +4,7 @@ package runtimechartemit
4
5 import (
6 "fmt"
7 + "maps"
8 "sort"
9 "strings"
10 "sync"
@@ -104,10 +105,15 @@ func (r *componentRegistry) snapshot() []componentSpec {
105
106 func cloneEmitEnv(env chartemit.EmitEnv) chartemit.EmitEnv {
107 out := env
107 - if env.JobLabels != nil {
108 - out.JobLabels = make(map[string]string, len(env.JobLabels))
109 - for k, v := range env.JobLabels {
110 - out.JobLabels[k] = v
108 + out.JobLabels = maps.Clone(env.JobLabels)
109 + if env.HostScope != nil {
110 + out.HostScope = &chartemit.HostScope{
111 + GUID: env.HostScope.GUID,
112 + }
113 + if env.HostScope.Define != nil {
114 + define := *env.HostScope.Define
115 + define.Labels = maps.Clone(env.HostScope.Define.Labels)
116 + out.HostScope.Define = &define
117 }
118 }
119 return out
@@ -143,7 +149,7 @@ func normalizeComponent(cfg ComponentConfig, pluginName string) (componentSpec,
149 Plugin: firstNotEmpty(strings.TrimSpace(cfg.Plugin), pluginName),
150 Module: firstNotEmpty(strings.TrimSpace(cfg.Module), "internal"),
151 JobName: firstNotEmpty(strings.TrimSpace(cfg.JobName), name),
146 - JobLabels: cloneStringMap(cfg.JobLabels),
152 + JobLabels: maps.Clone(cfg.JobLabels),
153 }
154
155 return componentSpec{
@@ -156,17 +162,6 @@ func normalizeComponent(cfg ComponentConfig, pluginName string) (componentSpec,
162 }, nil
163 }
164
159 -func cloneStringMap(in map[string]string) map[string]string {
160 - if len(in) == 0 {
161 - return nil
162 - }
163 - out := make(map[string]string, len(in))
164 - for k, v := range in {
165 - out[k] = v
166 - }
167 - return out
168 -}
169 -
165 func firstNotEmpty(items ...string) string {
166 for _, item := range items {
167 if item = strings.TrimSpace(item); item != "" {
src/go/plugin/agent/runtimechartemit/job.go
+188 -56
@@ -7,6 +7,7 @@ import (
7 "fmt"
8 "io"
9 "log/slog"
10 + "maps"
11 "sort"
12 "sync/atomic"
13 "time"
@@ -26,6 +27,46 @@ type runtimeComponentState struct {
27 knownCharts map[string]chartengine.ChartMeta
28 }
29
30 +type runtimePreparedEmission struct {
31 + attempt chartengine.PlanAttempt
32 + nextPrev time.Time
33 + nextKnownCharts map[string]chartengine.ChartMeta
34 +}
35 +
36 +type runtimeTickCommitStep struct {
37 + attempt chartengine.PlanAttempt
38 + finalize func()
39 +}
40 +
41 +type runtimeTickCommit struct {
42 + steps []runtimeTickCommitStep
43 +}
44 +
45 +func (c *runtimeTickCommit) add(step runtimeTickCommitStep) {
46 + c.steps = append(c.steps, step)
47 +}
48 +
49 +func (c *runtimeTickCommit) abort() {
50 + for _, step := range c.steps {
51 + step.attempt.Abort()
52 + }
53 +}
54 +
55 +func (c *runtimeTickCommit) commit() error {
56 + for i, step := range c.steps {
57 + if err := step.attempt.Commit(); err != nil {
58 + for _, remaining := range c.steps[i+1:] {
59 + remaining.attempt.Abort()
60 + }
61 + return err
62 + }
63 + if step.finalize != nil {
64 + step.finalize()
65 + }
66 + }
67 + return nil
68 +}
69 +
70 type runtimeMetricsJob struct {
71 *logger.Logger
72
@@ -123,6 +164,7 @@ func (j *runtimeMetricsJob) runOnce(clock int) {
164 specs := j.registry.snapshot()
165 seen := make(map[string]struct{}, len(specs))
166 now := time.Now()
167 + var commit runtimeTickCommit
168
169 for _, spec := range specs {
170 seen[spec.Name] = struct{}{}
@@ -131,59 +173,41 @@ func (j *runtimeMetricsJob) runOnce(clock int) {
173 continue
174 }
175
134 - component, err := j.ensureComponent(spec)
135 - if err != nil {
136 - j.Warningf("runtime metrics component %q init failed: %v", spec.Name, err)
137 - continue
138 - }
139 -
140 - reader := spec.Store.Read(metrix.ReadRaw(), metrix.ReadFlatten())
141 - plan, err := component.engine.BuildPlan(reader)
142 - if err != nil {
143 - j.Warningf("runtime metrics component %q build plan failed: %v", spec.Name, err)
176 + step, buf, ok := j.prepareComponentStep(spec, now)
177 + if !ok {
178 continue
179 }
146 - if len(plan.Actions) == 0 {
147 - continue
148 - }
149 -
150 - env := cloneEmitEnv(spec.EmitEnv)
151 - env.MSSinceLast = calcRuntimeSinceLast(now, component.prev)
152 - component.prev = now
153 -
154 - if err := chartemit.ApplyPlan(j.api, plan, env); err != nil {
155 - j.Warningf("runtime metrics component %q apply plan failed: %v", spec.Name, err)
156 - continue
180 + if buf != nil && buf.Len() > 0 {
181 + _, _ = j.buf.Write(buf.Bytes())
182 }
158 - component.trackPlan(plan)
183 + commit.add(step)
184 }
185
186 for name := range j.components {
162 - state, ok := j.components[name]
163 - if ok && state != nil {
164 - if _, exists := seen[name]; exists {
165 - continue
166 - }
167 - j.emitComponentObsolete(state)
168 - }
187 if _, ok := seen[name]; ok {
188 continue
189 }
172 - delete(j.components, name)
190 + state := j.components[name]
191 + step, buf, ok := j.prepareRemovalStep(name, state)
192 + if !ok {
193 + continue
194 + }
195 + if buf != nil && buf.Len() > 0 {
196 + _, _ = j.buf.Write(buf.Bytes())
197 + }
198 + commit.add(step)
199 }
200
201 if j.buf.Len() > 0 {
202 _, _ = io.Copy(j.out, j.buf)
177 - j.buf.Reset()
203 }
179 -}
180 -
181 -func (j *runtimeMetricsJob) ensureComponent(spec componentSpec) (*runtimeComponentState, error) {
182 - current, ok := j.components[spec.Name]
183 - if ok && current.spec.Generation == spec.Generation {
184 - return current, nil
204 + if err := commit.commit(); err != nil {
205 + j.Warningf("runtime metrics commit failed: %v", err)
206 }
207 + j.buf.Reset()
208 +}
209
210 +func (j *runtimeMetricsJob) newComponentState(spec componentSpec) (*runtimeComponentState, error) {
211 engineLog := j.Logger.With(slog.String("runtime_component", spec.Name))
212 engine, err := chartengine.New(
213 chartengine.WithRuntimeStore(nil), // Two-engine policy: observer engine has no self-metrics.
@@ -205,16 +229,110 @@ func (j *runtimeMetricsJob) ensureComponent(spec componentSpec) (*runtimeCompone
229 engine: engine,
230 knownCharts: make(map[string]chartengine.ChartMeta),
231 }
208 - if ok && current != nil && current.spec.Generation != spec.Generation {
209 - j.emitComponentObsolete(current)
210 - }
211 - j.components[spec.Name] = state
232 return state, nil
233 }
234
215 -func (j *runtimeMetricsJob) emitComponentObsolete(state *runtimeComponentState) {
235 +func (j *runtimeMetricsJob) prepareComponentStep(spec componentSpec, now time.Time) (runtimeTickCommitStep, *bytes.Buffer, bool) {
236 + current := j.components[spec.Name]
237 + if current != nil && current.spec.Generation == spec.Generation {
238 + emission, buf, ok := j.prepareEmissionToBuffer(current, now)
239 + if !ok {
240 + return runtimeTickCommitStep{}, nil, false
241 + }
242 + return runtimeTickCommitStep{
243 + attempt: emission.attempt,
244 + finalize: func() {
245 + current.prev = emission.nextPrev
246 + current.knownCharts = emission.nextKnownCharts
247 + },
248 + }, buf, true
249 + }
250 +
251 + next, err := j.newComponentState(spec)
252 + if err != nil {
253 + j.Warningf("runtime metrics component %q init failed: %v", spec.Name, err)
254 + return runtimeTickCommitStep{}, nil, false
255 + }
256 +
257 + var buf bytes.Buffer
258 + api := netdataapi.New(&buf)
259 + if err := j.emitComponentObsolete(api, current); err != nil {
260 + j.Warningf("runtime metrics component %q obsolete emit failed: %v", spec.Name, err)
261 + return runtimeTickCommitStep{}, nil, false
262 + }
263 + emission, ok := j.prepareEmission(api, next, now)
264 + if !ok {
265 + return runtimeTickCommitStep{}, nil, false
266 + }
267 + return runtimeTickCommitStep{
268 + attempt: emission.attempt,
269 + finalize: func() {
270 + next.prev = emission.nextPrev
271 + next.knownCharts = emission.nextKnownCharts
272 + j.components[spec.Name] = next
273 + },
274 + }, &buf, true
275 +}
276 +
277 +func (j *runtimeMetricsJob) prepareRemovalStep(name string, state *runtimeComponentState) (runtimeTickCommitStep, *bytes.Buffer, bool) {
278 + if state == nil {
279 + return runtimeTickCommitStep{}, nil, false
280 + }
281 +
282 + var buf bytes.Buffer
283 + api := netdataapi.New(&buf)
284 + if err := j.emitComponentObsolete(api, state); err != nil {
285 + j.Warningf("runtime metrics component %q obsolete emit failed: %v", state.spec.Name, err)
286 + return runtimeTickCommitStep{}, nil, false
287 + }
288 + return runtimeTickCommitStep{
289 + finalize: func() {
290 + delete(j.components, name)
291 + },
292 + }, &buf, true
293 +}
294 +
295 +func (j *runtimeMetricsJob) prepareEmissionToBuffer(state *runtimeComponentState, now time.Time) (runtimePreparedEmission, *bytes.Buffer, bool) {
296 + var buf bytes.Buffer
297 + api := netdataapi.New(&buf)
298 + emission, ok := j.prepareEmission(api, state, now)
299 + if !ok {
300 + return runtimePreparedEmission{}, nil, false
301 + }
302 + return emission, &buf, true
303 +}
304 +
305 +func (j *runtimeMetricsJob) prepareEmission(api *netdataapi.API, state *runtimeComponentState, now time.Time) (runtimePreparedEmission, bool) {
306 + if state == nil {
307 + return runtimePreparedEmission{}, false
308 + }
309 +
310 + reader := state.spec.Store.Read(metrix.ReadRaw(), metrix.ReadFlatten())
311 + attempt, err := state.engine.PreparePlan(reader)
312 + if err != nil {
313 + j.Warningf("runtime metrics component %q build plan failed: %v", state.spec.Name, err)
314 + return runtimePreparedEmission{}, false
315 + }
316 +
317 + plan := attempt.Plan()
318 + env := cloneEmitEnv(state.spec.EmitEnv)
319 + env.MSSinceLast = calcRuntimeSinceLast(now, state.prev)
320 + if err := chartemit.ApplyPlan(api, plan, env); err != nil {
321 + attempt.Abort()
322 + j.Warningf("runtime metrics component %q apply plan failed: %v", state.spec.Name, err)
323 + return runtimePreparedEmission{}, false
324 + }
325 +
326 + return runtimePreparedEmission{
327 + attempt: attempt,
328 + nextPrev: now,
329 + nextKnownCharts: applyEffectiveChartSet(state.knownCharts, plan),
330 + }, true
331 +}
332 +
333 +func (j *runtimeMetricsJob) emitComponentObsolete(api *netdataapi.API, state *runtimeComponentState) error {
334 if state == nil || len(state.knownCharts) == 0 {
217 - return
335 + return nil
336 }
337
338 chartIDs := make([]string, 0, len(state.knownCharts))
@@ -234,28 +352,42 @@ func (j *runtimeMetricsJob) emitComponentObsolete(state *runtimeComponentState)
352
353 env := cloneEmitEnv(state.spec.EmitEnv)
354 env.MSSinceLast = 0
237 - if err := chartemit.ApplyPlan(j.api, chartengine.Plan{Actions: actions}, env); err != nil {
238 - j.Warningf("runtime metrics component %q obsolete emit failed: %v", state.spec.Name, err)
239 - return
240 - }
241 - clear(state.knownCharts)
355 + return chartemit.ApplyPlan(api, chartengine.Plan{Actions: actions}, env)
356 }
357
244 -func (s *runtimeComponentState) trackPlan(plan chartengine.Plan) {
245 - if s == nil {
246 - return
247 - }
248 - if s.knownCharts == nil {
249 - s.knownCharts = make(map[string]chartengine.ChartMeta)
358 +func applyEffectiveChartSet(known map[string]chartengine.ChartMeta, plan chartengine.Plan) map[string]chartengine.ChartMeta {
359 + out := maps.Clone(known)
360 + if out == nil {
361 + out = make(map[string]chartengine.ChartMeta)
362 }
363 +
364 + createCharts := make(map[string]chartengine.ChartMeta)
365 + dimensionOnlyCharts := make(map[string]chartengine.ChartMeta)
366 for _, action := range plan.Actions {
367 switch v := action.(type) {
368 case chartengine.CreateChartAction:
254 - s.knownCharts[v.ChartID] = v.Meta
369 + createCharts[v.ChartID] = v.Meta
370 + case chartengine.CreateDimensionAction:
371 + if _, ok := createCharts[v.ChartID]; ok {
372 + continue
373 + }
374 + if _, ok := dimensionOnlyCharts[v.ChartID]; !ok {
375 + dimensionOnlyCharts[v.ChartID] = v.ChartMeta
376 + }
377 case chartengine.RemoveChartAction:
256 - delete(s.knownCharts, v.ChartID)
378 + delete(out, v.ChartID)
379 + }
380 + }
381 + for chartID, meta := range createCharts {
382 + out[chartID] = meta
383 + }
384 + for chartID, meta := range dimensionOnlyCharts {
385 + if _, ok := out[chartID]; ok {
386 + continue
387 }
388 + out[chartID] = meta
389 }
390 + return out
391 }
392
393 func calcRuntimeSinceLast(cur, prev time.Time) int {
src/go/plugin/agent/runtimechartemit/job_test.go
+187 -1
@@ -4,11 +4,14 @@ package runtimechartemit
4
5 import (
6 "bytes"
7 + "errors"
8 + "strings"
9 "sync"
10 "testing"
11 "time"
12
13 "github.com/netdata/netdata/go/plugins/plugin/framework/chartemit"
14 + "github.com/netdata/netdata/go/plugins/plugin/framework/chartengine"
15 "github.com/stretchr/testify/assert"
16 "github.com/stretchr/testify/require"
17
@@ -38,6 +41,23 @@ func (b *safeBuffer) String() string {
41 return b.buf.String()
42 }
43
44 +type failingWriter struct{}
45 +
46 +func (failingWriter) Write(_ []byte) (int, error) {
47 + return 0, errors.New("boom")
48 +}
49 +
50 +func requireInOrder(t *testing.T, text string, parts ...string) {
51 + t.Helper()
52 +
53 + offset := 0
54 + for _, part := range parts {
55 + idx := strings.Index(text[offset:], part)
56 + require.NotEqualf(t, -1, idx, "missing ordered fragment %q in %q", part, text)
57 + offset += idx + len(part)
58 + }
59 +}
60 +
61 func TestRuntimeMetricsJobStartStopLifecycle(t *testing.T) {
62 tests := map[string]struct {
63 clock int
@@ -89,7 +109,7 @@ func TestRuntimeMetricsJobStartStopLifecycle(t *testing.T) {
109 }
110
111 assert.False(t, job.running.Load())
92 - assert.Contains(t, out.String(), "BEGIN")
112 + requireInOrder(t, out.String(), "HOST ''", "BEGIN")
113 })
114 }
115 }
@@ -121,3 +141,169 @@ func TestRuntimeMetricsJobTickSkipWhenBusy(t *testing.T) {
141 })
142 }
143 }
144 +
145 +func TestRuntimeMetricsJobTransactionalScenarios(t *testing.T) {
146 + tests := map[string]struct {
147 + run func(t *testing.T)
148 + }{
149 + "flush failure does not block component state advancement": {
150 + run: func(t *testing.T) {
151 + reg := newComponentRegistry()
152 + store := metrix.NewRuntimeStore()
153 + store.Write().StatefulMeter("component").Gauge("load").Set(5)
154 +
155 + reg.upsert(componentSpec{
156 + Name: "component",
157 + Store: store,
158 + TemplateYAML: []byte(runtimeGaugeTemplateYAML()),
159 + UpdateEvery: 1,
160 + EmitEnv: chartemit.EmitEnv{
161 + TypeID: "netdata.go.d.internal.component",
162 + UpdateEvery: 1,
163 + Plugin: "go.d",
164 + Module: "internal",
165 + JobName: "component",
166 + },
167 + })
168 +
169 + job := newRuntimeMetricsJob(failingWriter{}, reg, nil)
170 + job.runOnce(1)
171 +
172 + state := job.components["component"]
173 + require.NotNil(t, state)
174 + require.False(t, state.prev.IsZero())
175 + require.NotEmpty(t, state.knownCharts)
176 +
177 + var out safeBuffer
178 + job.out = &out
179 + job.runOnce(2)
180 +
181 + state = job.components["component"]
182 + require.NotNil(t, state)
183 + require.False(t, state.prev.IsZero())
184 + require.NotEmpty(t, state.knownCharts)
185 + requireInOrder(t, out.String(), "HOST ''", "BEGIN")
186 + assert.NotContains(t, out.String(), "CHART 'component_load'")
187 + },
188 + },
189 + "effective chart tracking includes dimension only creation": {
190 + run: func(t *testing.T) {
191 + plan := chartengine.Plan{
192 + Actions: []chartengine.EngineAction{
193 + chartengine.CreateDimensionAction{
194 + ChartID: "component_load",
195 + ChartMeta: chartengine.ChartMeta{
196 + Title: "Component Load",
197 + Context: "netdata.go.plugin.component.component_load",
198 + Units: "load",
199 + },
200 + Name: "value",
201 + },
202 + },
203 + }
204 +
205 + known := applyEffectiveChartSet(nil, plan)
206 + require.Contains(t, known, "component_load")
207 +
208 + known = applyEffectiveChartSet(known, chartengine.Plan{
209 + Actions: []chartengine.EngineAction{
210 + chartengine.RemoveChartAction{
211 + ChartID: "component_load",
212 + Meta: chartengine.ChartMeta{
213 + Title: "Component Load",
214 + Context: "netdata.go.plugin.component.component_load",
215 + Units: "load",
216 + },
217 + },
218 + },
219 + })
220 + require.NotContains(t, known, "component_load")
221 + },
222 + },
223 + "generation replacement keeps old state when obsolete emit fails": {
224 + run: func(t *testing.T) {
225 + reg := newComponentRegistry()
226 + store := metrix.NewRuntimeStore()
227 + store.Write().StatefulMeter("component").Gauge("load").Set(7)
228 + reg.upsert(componentSpec{
229 + Name: "component",
230 + Store: store,
231 + TemplateYAML: []byte(runtimeGaugeTemplateYAML()),
232 + UpdateEvery: 1,
233 + EmitEnv: chartemit.EmitEnv{
234 + TypeID: "netdata.go.d.internal.component",
235 + UpdateEvery: 1,
236 + Plugin: "go.d",
237 + Module: "internal",
238 + JobName: "component",
239 + },
240 + })
241 +
242 + job := newRuntimeMetricsJob(&safeBuffer{}, reg, nil)
243 + current := &runtimeComponentState{
244 + spec: componentSpec{
245 + Name: "component",
246 + Generation: 0,
247 + EmitEnv: chartemit.EmitEnv{
248 + TypeID: " ",
249 + UpdateEvery: 1,
250 + Plugin: "go.d",
251 + Module: "internal",
252 + JobName: "component",
253 + },
254 + },
255 + prev: time.Unix(1, 0),
256 + knownCharts: map[string]chartengine.ChartMeta{
257 + "component_load": {
258 + Title: "Component Load",
259 + Context: "netdata.go.plugin.component.component_load",
260 + Units: "load",
261 + },
262 + },
263 + }
264 + job.components["component"] = current
265 +
266 + job.runOnce(1)
267 +
268 + require.Same(t, current, job.components["component"])
269 + assert.Equal(t, uint64(0), job.components["component"].spec.Generation)
270 + },
271 + },
272 + "removal keeps old state when obsolete emit fails": {
273 + run: func(t *testing.T) {
274 + job := newRuntimeMetricsJob(&safeBuffer{}, newComponentRegistry(), nil)
275 + current := &runtimeComponentState{
276 + spec: componentSpec{
277 + Name: "component",
278 + Generation: 1,
279 + EmitEnv: chartemit.EmitEnv{
280 + TypeID: " ",
281 + UpdateEvery: 1,
282 + Plugin: "go.d",
283 + Module: "internal",
284 + JobName: "component",
285 + },
286 + },
287 + prev: time.Unix(1, 0),
288 + knownCharts: map[string]chartengine.ChartMeta{
289 + "component_load": {
290 + Title: "Component Load",
291 + Context: "netdata.go.plugin.component.component_load",
292 + Units: "load",
293 + },
294 + },
295 + }
296 + job.components["component"] = current
297 +
298 + job.runOnce(1)
299 +
300 + require.Same(t, current, job.components["component"])
301 + require.Contains(t, job.components["component"].knownCharts, "component_load")
302 + },
303 + },
304 + }
305 +
306 + for name, tc := range tests {
307 + t.Run(name, tc.run)
308 + }
309 +}
src/go/plugin/agent/runtimechartemit/service_test.go
+2 -4
@@ -117,8 +117,7 @@ func TestRuntimeMetricsJobScenarios(t *testing.T) {
117
118 job.runOnce(2)
119 result := out.String()
120 - assert.Contains(t, result, "CHART")
121 - assert.Contains(t, result, "BEGIN")
120 + requireInOrder(t, result, "HOST ''", "CHART", "BEGIN")
121 },
122 },
123 "runtime job observes all visible runtime series (not only latest seq)": {
@@ -211,8 +210,7 @@ func TestRuntimeMetricsJobScenarios(t *testing.T) {
210 reg.remove("component")
211 job.runOnce(2)
212 result := out.String()
214 - assert.Contains(t, result, "CHART 'netdata.go.d.internal.component.component_load'")
215 - assert.Contains(t, result, "'obsolete'")
213 + requireInOrder(t, result, "HOST ''", "CHART 'netdata.go.d.internal.component.component_load'", "'obsolete'")
214 },
215 },
216 }
src/go/plugin/framework/chartemit/apply.go
+41
@@ -51,12 +51,53 @@ func ApplyPlan(api *netdataapi.API, plan Plan, env EmitEnv) error {
51 if err := validateTypeIDBudget(env.TypeID, normalized); err != nil {
52 return err
53 }
54 + if !hasEmissions(normalized) {
55 + return nil
56 + }
57 + if err := emitHostSelection(api, env); err != nil {
58 + return err
59 + }
60 emitCreatePhase(api, env, normalized)
61 emitUpdatePhase(api, env, normalized.updateCharts)
62 emitRemovePhase(api, env, normalized)
63 return nil
64 }
65
66 +func hasEmissions(actions normalizedActions) bool {
67 + return len(actions.createCharts) > 0 ||
68 + len(actions.createDimsByID) > 0 ||
69 + len(actions.updateCharts) > 0 ||
70 + len(actions.removeDimensions) > 0 ||
71 + len(actions.removeCharts) > 0
72 +}
73 +
74 +func emitHostSelection(api *netdataapi.API, env EmitEnv) error {
75 + if env.HostScope == nil {
76 + api.HOST("")
77 + return nil
78 + }
79 +
80 + guid := sanitizeWireID(env.HostScope.GUID)
81 + if guid == "" {
82 + return fmt.Errorf("chartemit: emit env host scope guid is required")
83 + }
84 + if env.HostScope.Define != nil {
85 + defineGUID := sanitizeWireID(env.HostScope.Define.GUID)
86 + if defineGUID == "" {
87 + return fmt.Errorf("chartemit: host define guid is required")
88 + }
89 + if defineGUID != guid {
90 + return fmt.Errorf("chartemit: host define guid %q does not match host scope guid %q", env.HostScope.Define.GUID, env.HostScope.GUID)
91 + }
92 + if strings.TrimSpace(env.HostScope.Define.Hostname) == "" {
93 + return fmt.Errorf("chartemit: host define hostname is required")
94 + }
95 + api.HOSTINFO(*env.HostScope.Define)
96 + }
97 + api.HOST(guid)
98 + return nil
99 +}
100 +
101 func validateTypeIDBudget(typeID string, actions normalizedActions) error {
102 typeID = sanitizeWireID(typeID)
103 if typeID == "" {
src/go/plugin/framework/chartemit/apply_test.go
+240 -37
@@ -4,7 +4,6 @@ package chartemit
4
5 import (
6 "bytes"
7 - "strings"
7 "testing"
8
9 "github.com/netdata/netdata/go/plugins/plugin/framework/chartengine"
@@ -86,21 +85,21 @@ func TestApplyPlanEmitsNetdataWire(t *testing.T) {
85 require.NoError(t, err)
86
87 out := buf.String()
89 - assert.Contains(t, out, "CHART 'collector.job.win_nic_traffic_eth0'")
90 - assert.Contains(t, out, "CLABEL 'instance' 'localhost' '2'")
91 - assert.Contains(t, out, "CLABEL '_collect_job' 'job01' '1'")
92 - assert.Contains(t, out, "CLABEL_COMMIT")
93 - assert.Contains(t, out, "DIMENSION 'received' 'received' 'incremental' '1' '1' 'type=float'")
94 - assert.Contains(t, out, "BEGIN 'collector.job.win_nic_traffic_eth0' 100")
95 - assert.Contains(t, out, "SET 'received' = 123.5")
96 - assert.Contains(t, out, "DIMENSION 'received' 'received' 'incremental' '1' '1' 'obsolete type=float'")
97 - assert.Contains(t, out, "obsolete")
98 -
99 - createPos := strings.Index(out, "CHART 'collector.job.win_nic_traffic_eth0'")
100 - beginPos := strings.Index(out, "BEGIN 'collector.job.win_nic_traffic_eth0' 100")
101 - require.NotEqual(t, -1, createPos)
102 - require.NotEqual(t, -1, beginPos)
103 - assert.Less(t, createPos, beginPos)
88 + assert.Equal(t, `HOST ''
89 +
90 +CHART 'collector.job.win_nic_traffic_eth0' '' 'NIC traffic' 'bytes/s' 'Net' 'nic_traffic' 'line' '1' '5' '' 'go.d.plugin' 'windows'
91 +CLABEL 'instance' 'localhost' '2'
92 +CLABEL '_collect_job' 'job01' '1'
93 +CLABEL_COMMIT
94 +DIMENSION 'received' 'received' 'incremental' '1' '1' 'type=float'
95 +BEGIN 'collector.job.win_nic_traffic_eth0' 100
96 +SET 'received' = 123.5
97 +END
98 +
99 +CHART 'collector.job.win_nic_traffic_eth0' '' 'NIC traffic' 'bytes/s' 'Net' 'nic_traffic' 'line' '1' '5' '' 'go.d.plugin' 'windows'
100 +DIMENSION 'received' 'received' 'incremental' '1' '1' 'obsolete type=float'
101 +CHART 'collector.job.win_nic_traffic_eth0' '' 'NIC traffic' 'bytes/s' 'Net' 'nic_traffic' 'line' '1' '5' 'obsolete' 'go.d.plugin' 'windows'
102 +`, out)
103 }
104
105 func TestApplyPlanAutogenChartCreateUpdateRemove(t *testing.T) {
@@ -161,12 +160,19 @@ func TestApplyPlanAutogenChartCreateUpdateRemove(t *testing.T) {
160 require.NoError(t, err)
161
162 out := buf.String()
164 - assert.Contains(t, out, "CHART 'collector.job.svc.errors_total-method=GET'")
165 - assert.Contains(t, out, "CLABEL 'method' 'GET' '1'")
166 - assert.Contains(t, out, "DIMENSION 'svc.errors_total' 'svc.errors_total' 'incremental' '1' '1' ''")
167 - assert.Contains(t, out, "BEGIN 'collector.job.svc.errors_total-method=GET'")
168 - assert.Contains(t, out, "SET 'svc.errors_total' = 10")
169 - assert.Contains(t, out, "obsolete")
163 + assert.Equal(t, `HOST ''
164 +
165 +CHART 'collector.job.svc.errors_total-method=GET' '' 'Metric "svc.errors_total"' 'events/s' 'svc_errors' 'svc.errors_total' 'line' '0' '1' '' 'go.d.plugin' 'prometheus'
166 +CLABEL 'method' 'GET' '1'
167 +CLABEL '_collect_job' 'job01' '1'
168 +CLABEL_COMMIT
169 +DIMENSION 'svc.errors_total' 'svc.errors_total' 'incremental' '1' '1' ''
170 +BEGIN 'collector.job.svc.errors_total-method=GET'
171 +SET 'svc.errors_total' = 10
172 +END
173 +
174 +CHART 'collector.job.svc.errors_total-method=GET' '' 'Metric "svc.errors_total"' 'events/s' 'svc_errors' 'svc.errors_total' 'line' '0' '1' 'obsolete' 'go.d.plugin' 'prometheus'
175 +`, out)
176 }
177
178 func TestApplyPlanUsesIntegerSETForNonFloatUpdates(t *testing.T) {
@@ -222,9 +228,17 @@ func TestApplyPlanUsesIntegerSETForNonFloatUpdates(t *testing.T) {
228 require.NoError(t, err)
229
230 out := buf.String()
225 - assert.Contains(t, out, "DIMENSION 'total' 'total' 'absolute' '1' '1' ''")
226 - assert.Contains(t, out, "BEGIN 'collector.job.runtime_jobs' 1")
227 - assert.Contains(t, out, "SET 'total' = 7")
231 + assert.Equal(t, `HOST ''
232 +
233 +CHART 'collector.job.runtime_jobs' '' 'Runtime jobs' 'jobs' 'Runtime' 'runtime.jobs' 'line' '0' '1' '' 'go.d.plugin' 'runtime'
234 +CLABEL '_collect_job' 'job01' '1'
235 +CLABEL_COMMIT
236 +DIMENSION 'total' 'total' 'absolute' '1' '1' ''
237 +BEGIN 'collector.job.runtime_jobs' 1
238 +SET 'total' = 7
239 +END
240 +
241 +`, out)
242 assert.NotContains(t, out, "SET 'total' = 7.9")
243 }
244
@@ -267,11 +281,14 @@ func TestApplyPlanDimensionOnlyCreateEmitsLabelsAndCommit(t *testing.T) {
281 require.NoError(t, err)
282
283 out := buf.String()
270 - assert.Contains(t, out, "CHART 'collector.job.dimension_only_chart'")
271 - assert.Contains(t, out, "CLABEL 'instance' 'localhost' '2'")
272 - assert.Contains(t, out, "CLABEL '_collect_job' 'job01' '1'")
273 - assert.Contains(t, out, "CLABEL_COMMIT")
274 - assert.Contains(t, out, "DIMENSION 'value' 'value' 'absolute' '1' '1' ''")
284 + assert.Equal(t, `HOST ''
285 +
286 +CHART 'collector.job.dimension_only_chart' '' 'Dimension-only chart' '1' 'Runtime' 'runtime.dimension_only' 'line' '0' '1' '' 'go.d.plugin' 'runtime'
287 +CLABEL 'instance' 'localhost' '2'
288 +CLABEL '_collect_job' 'job01' '1'
289 +CLABEL_COMMIT
290 +DIMENSION 'value' 'value' 'absolute' '1' '1' ''
291 +`, out)
292 }
293
294 func TestApplyPlanSanitizesWireValues(t *testing.T) {
@@ -332,13 +349,19 @@ func TestApplyPlanSanitizesWireValues(t *testing.T) {
349 require.NoError(t, err)
350
351 out := buf.String()
335 - assert.Contains(t, out, "CHART 'collector.job.chartid'")
336 - assert.Contains(t, out, "DIMENSION 'dimname' 'dimname' 'absolute' '1' '1' ''")
337 - assert.Contains(t, out, "BEGIN 'collector.job.chartid' 1")
338 - assert.Contains(t, out, "SET 'dimname' = 5")
339 - assert.Contains(t, out, "CLABEL 'instance' 'localhost ' '2'")
340 - assert.Contains(t, out, "CLABEL 'label' 'value ' '1'")
341 - assert.Contains(t, out, "CLABEL '_collect_job' 'job01 ' '1'")
352 + assert.Equal(t, `HOST ''
353 +
354 +CHART 'collector.job.chartid' '' 'Title ' 'units ' 'Family ' 'Context ' 'line' '0' '1' '' 'go.dplugin ' 'module '
355 +CLABEL 'instance' 'localhost ' '2'
356 +CLABEL 'label' 'value ' '1'
357 +CLABEL '_collect_job' 'job01 ' '1'
358 +CLABEL_COMMIT
359 +DIMENSION 'dimname' 'dimname' 'absolute' '1' '1' ''
360 +BEGIN 'collector.job.chartid' 1
361 +SET 'dimname' = 5
362 +END
363 +
364 +`, out)
365 }
366
367 func TestApplyPlanRejectsEmptyTypeID(t *testing.T) {
@@ -370,6 +393,186 @@ func TestApplyPlanRejectsEmptyTypeID(t *testing.T) {
393 }
394 }
395
396 +func TestApplyPlanDefaultGlobalHostSelection(t *testing.T) {
397 + var buf bytes.Buffer
398 + api := netdataapi.New(&buf)
399 +
400 + meta := chartengine.ChartMeta{
401 + Title: "Requests",
402 + Family: "Service",
403 + Context: "requests",
404 + Units: "req/s",
405 + Type: chartengine.ChartTypeLine,
406 + }
407 + plan := Plan{
408 + Actions: []EngineAction{
409 + chartengine.CreateChartAction{
410 + ChartID: "requests",
411 + Meta: meta,
412 + },
413 + },
414 + }
415 +
416 + require.NoError(t, ApplyPlan(api, plan, EmitEnv{
417 + TypeID: "collector.job",
418 + UpdateEvery: 1,
419 + Plugin: "go.d.plugin",
420 + Module: "httpcheck",
421 + JobName: "job01",
422 + }))
423 +
424 + out := buf.String()
425 + assert.Equal(t, `HOST ''
426 +
427 +CHART 'collector.job.requests' '' 'Requests' 'req/s' 'Service' 'requests' 'line' '0' '1' '' 'go.d.plugin' 'httpcheck'
428 +CLABEL '_collect_job' 'job01' '1'
429 +CLABEL_COMMIT
430 +`, out)
431 +}
432 +
433 +func TestApplyPlanVnodeHostSelectionAndDefine(t *testing.T) {
434 + var buf bytes.Buffer
435 + api := netdataapi.New(&buf)
436 +
437 + meta := chartengine.ChartMeta{
438 + Title: "Workers Busy",
439 + Family: "Workers",
440 + Context: "workers_busy",
441 + Units: "workers",
442 + Type: chartengine.ChartTypeLine,
443 + }
444 + info, err := PrepareHostInfo(netdataapi.HostInfo{
445 + GUID: "node-guid",
446 + Hostname: "node-host",
447 + Labels: map[string]string{
448 + "region": "eu'\n",
449 + },
450 + })
451 + require.NoError(t, err)
452 +
453 + plan := Plan{
454 + Actions: []EngineAction{
455 + chartengine.CreateChartAction{
456 + ChartID: "workers_busy",
457 + Meta: meta,
458 + },
459 + },
460 + }
461 +
462 + require.NoError(t, ApplyPlan(api, plan, EmitEnv{
463 + TypeID: "collector.job",
464 + UpdateEvery: 1,
465 + Plugin: "go.d.plugin",
466 + Module: "apache",
467 + JobName: "job01",
468 + HostScope: &HostScope{
469 + GUID: "node-guid",
470 + Define: &info,
471 + },
472 + }))
473 +
474 + out := buf.String()
475 + assert.Equal(t, `HOST_DEFINE 'node-guid' 'node-host'
476 +HOST_LABEL '_hostname' 'node-host'
477 +HOST_LABEL 'region' 'eu '
478 +HOST_DEFINE_END
479 +
480 +HOST 'node-guid'
481 +
482 +CHART 'collector.job.workers_busy' '' 'Workers Busy' 'workers' 'Workers' 'workers_busy' 'line' '0' '1' '' 'go.d.plugin' 'apache'
483 +CLABEL '_collect_job' 'job01' '1'
484 +CLABEL_COMMIT
485 +`, out)
486 +}
487 +
488 +func TestApplyPlanSkipsHostSelectionForEmptyPlans(t *testing.T) {
489 + var buf bytes.Buffer
490 + api := netdataapi.New(&buf)
491 +
492 + require.NoError(t, ApplyPlan(api, Plan{}, EmitEnv{
493 + TypeID: "collector.job",
494 + UpdateEvery: 1,
495 + Plugin: "go.d.plugin",
496 + Module: "runtime",
497 + JobName: "job01",
498 + }))
499 + assert.Equal(t, "", buf.String())
500 +}
501 +
502 +func TestApplyPlanRejectsMismatchedHostDefine(t *testing.T) {
503 + var buf bytes.Buffer
504 + api := netdataapi.New(&buf)
505 +
506 + meta := chartengine.ChartMeta{
507 + Title: "Requests",
508 + Family: "Service",
509 + Context: "requests",
510 + Units: "req/s",
511 + Type: chartengine.ChartTypeLine,
512 + }
513 + plan := Plan{
514 + Actions: []EngineAction{
515 + chartengine.CreateChartAction{
516 + ChartID: "requests",
517 + Meta: meta,
518 + },
519 + },
520 + }
521 +
522 + err := ApplyPlan(api, plan, EmitEnv{
523 + TypeID: "collector.job",
524 + UpdateEvery: 1,
525 + Plugin: "go.d.plugin",
526 + Module: "httpcheck",
527 + JobName: "job01",
528 + HostScope: &HostScope{
529 + GUID: "guid-a",
530 + Define: &netdataapi.HostInfo{
531 + GUID: "guid-b",
532 + Hostname: "node-host",
533 + },
534 + },
535 + })
536 + require.Error(t, err)
537 + assert.ErrorContains(t, err, "does not match")
538 + assert.Equal(t, "", buf.String())
539 +}
540 +
541 +func TestApplyPlanRemoveOnlyBatchStillSelectsHost(t *testing.T) {
542 + var buf bytes.Buffer
543 + api := netdataapi.New(&buf)
544 +
545 + meta := chartengine.ChartMeta{
546 + Title: "Requests",
547 + Family: "Service",
548 + Context: "requests",
549 + Units: "req/s",
550 + Type: chartengine.ChartTypeLine,
551 + }
552 + plan := Plan{
553 + Actions: []EngineAction{
554 + chartengine.RemoveChartAction{
555 + ChartID: "requests",
556 + Meta: meta,
557 + },
558 + },
559 + }
560 +
561 + require.NoError(t, ApplyPlan(api, plan, EmitEnv{
562 + TypeID: "collector.job",
563 + UpdateEvery: 1,
564 + Plugin: "go.d.plugin",
565 + Module: "httpcheck",
566 + JobName: "job01",
567 + }))
568 +
569 + out := buf.String()
570 + assert.Equal(t, `HOST ''
571 +
572 +CHART 'collector.job.requests' '' 'Requests' 'req/s' 'Service' 'requests' 'line' '0' '1' 'obsolete' 'go.d.plugin' 'httpcheck'
573 +`, out)
574 +}
575 +
576 func TestNormalizeActionsOrderingDeterminism(t *testing.T) {
577 meta := chartengine.ChartMeta{
578 Title: "Requests",
src/go/plugin/framework/chartemit/host.go new
+45
@@ -0,0 +1,45 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package chartemit
4 +
5 +import (
6 + "fmt"
7 + "maps"
8 + "strings"
9 +
10 + "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
11 +)
12 +
13 +// PrepareHostInfo normalizes host-definition payloads before HOST_DEFINE.
14 +//
15 +// Current semantics intentionally match v1 vnode emission:
16 +// - GUID/hostname must be present,
17 +// - "_hostname" is injected when absent,
18 +// - label values are sanitized for Netdata wire output.
19 +func PrepareHostInfo(info netdataapi.HostInfo) (netdataapi.HostInfo, error) {
20 + guid := strings.TrimSpace(info.GUID)
21 + if guid == "" {
22 + return netdataapi.HostInfo{}, fmt.Errorf("chartemit: host guid is required")
23 + }
24 + hostname := strings.TrimSpace(info.Hostname)
25 + if hostname == "" {
26 + return netdataapi.HostInfo{}, fmt.Errorf("chartemit: host hostname is required")
27 + }
28 +
29 + labels := maps.Clone(info.Labels)
30 + if labels == nil {
31 + labels = make(map[string]string)
32 + }
33 + if _, ok := labels["_hostname"]; !ok {
34 + labels["_hostname"] = hostname
35 + }
36 + for key, value := range labels {
37 + labels[key] = sanitizeWireValue(value)
38 + }
39 +
40 + return netdataapi.HostInfo{
41 + GUID: guid,
42 + Hostname: hostname,
43 + Labels: labels,
44 + }, nil
45 +}
src/go/plugin/framework/chartemit/types.go
+11
@@ -3,9 +3,19 @@
3 package chartemit
4
5 import (
6 + "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
7 "github.com/netdata/netdata/go/plugins/plugin/framework/chartengine"
8 )
9
10 +// HostScope controls which Netdata host context one emitted chart batch uses.
11 +//
12 +// Nil means "explicit global context". Define is optional and caller-driven:
13 +// callers decide when a host still needs HOST_DEFINE before selecting it.
14 +type HostScope struct {
15 + GUID string
16 + Define *netdataapi.HostInfo
17 +}
18 +
19 // EmitEnv carries runtime context for translating engine actions to Netdata wire.
20 type EmitEnv struct {
21 TypeID string
@@ -14,6 +24,7 @@ type EmitEnv struct {
24 Module string
25 JobName string
26 JobLabels map[string]string
27 + HostScope *HostScope
28 MSSinceLast int
29 }
30
src/go/plugin/framework/chartengine/README.md
+26 -21
@@ -9,12 +9,12 @@
9
10 ## Purpose
11
12 -| Stage | Responsibility |
13 -|-----------------------|-------------------------------------------|
14 -| `charttpl` | Template decode/defaults/validation |
15 -| `chartengine.Compile` | Build immutable program IR |
16 -| `Engine.BuildPlan` | Produce plan actions from metric snapshot |
17 -| `chartemit.ApplyPlan` | Emit plan to Netdata wire protocol |
12 +| Stage | Responsibility |
13 +|-----------------------|----------------------------------------------------------|
14 +| `charttpl` | Template decode/defaults/validation |
15 +| `chartengine.Compile` | Build immutable program IR |
16 +| `Engine.PreparePlan` | Prepare plan actions plus explicit commit/abort boundary |
17 +| `chartemit.ApplyPlan` | Emit plan to Netdata wire protocol |
18
19 ## Collector-Facing Contract
20
@@ -33,7 +33,7 @@ For `ModuleV2` collectors, the runtime integration expects:
33 |-----------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
34 | `New(opts...)` | Create engine with policy/runtime options |
35 | `Load(spec, revision)` / `LoadYAML(data, revision)` | Compile and publish program revision |
36 -| `BuildPlan(reader)` | Build deterministic action plan from reader snapshot |
36 +| `PreparePlan(reader)` | Build deterministic action plan from reader snapshot and return an explicit attempt |
37 | `RuntimeStore()` | Access chartengine internal runtime metrics store |
38 | `WithEnginePolicy(...)` | Configure selector + autogen behavior |
39 | `WithRuntimeStore(...)` | Override/disable self-metrics store |
@@ -51,9 +51,9 @@ meter.Counter("requests_total").ObserveTotal(100)
51
52 // 2) Engine loads chart template.
53 engine, err := chartengine.New(
54 - chartengine.WithEnginePolicy(chartengine.EnginePolicy{
55 - Autogen: &chartengine.AutogenPolicy{Enabled: false},
56 - }),
54 + chartengine.WithEnginePolicy(chartengine.EnginePolicy{
55 + Autogen: &chartengine.AutogenPolicy{Enabled: false},
56 + }),
57 )
58 // handle err
59
@@ -73,12 +73,14 @@ groups:
73 `), 1)
74 // handle err
75
76 -// 3) Build plan from flattened+raw reader and emit.
76 +// 3) Prepare plan from flattened+raw reader, emit it, then commit.
77 // ReadFlatten() is included even for templates with static dimensions
78 // because it is required for inferred dimensions, structured-family autogen,
79 // and is the standard pattern.
80 -plan, err := engine.BuildPlan(store.Read(metrix.ReadRaw(), metrix.ReadFlatten()))
80 +attempt, err := engine.PreparePlan(store.Read(metrix.ReadRaw(), metrix.ReadFlatten()))
81 // handle err
82 +plan := attempt.Plan()
83 +defer attempt.Abort()
84
85 err = chartemit.ApplyPlan(api, plan, chartemit.EmitEnv{
86 TypeID: "plugin.job",
@@ -88,11 +90,14 @@ Module: "example",
90 JobName: "example",
91 })
92 // handle err
93 +err = attempt.Commit()
94 +// handle err
95 +
96 ```
97
93 -## BuildPlan Lifecycle
98 +## PreparePlan Lifecycle
99
95 -`BuildPlan` executes a deterministic phase pipeline.
100 +`PreparePlan` executes a deterministic phase pipeline.
101 Terms like "materialized state" and "route cache" are defined in the Engine State section below.
102
103 | Phase | Summary |
@@ -115,7 +120,7 @@ Terms like "materialized state" and "route cache" are defined in the Engine Stat
120 | Structured autogen families (`Histogram`, `Summary`, `StateSet`, `MeasureSet`) | Must use flattened reader metadata (`ReadFlatten`) or they are not visible |
121 | Runtime/default `ModuleV2` path | `Read(ReadRaw(), ReadFlatten())` |
122
118 -If inferred dimensions are present without flattened reader metadata, `BuildPlan` returns an explicit error.
123 +If inferred dimensions are present without flattened reader metadata, `PreparePlan` returns an explicit error.
124
125 ## Action Semantics
126
@@ -183,12 +188,12 @@ Default lifecycle policy when template omits lifecycle:
188
189 These label keys are treated specially by chartengine when consuming flattened structured-family or distribution inputs:
190
186 -| Key / Pattern | Meaning |
187 -|-------------------|---------|
188 -| `le` | Histogram bucket bound label |
189 -| `quantile` | Summary quantile label |
190 -| `measure_field` | `MeasureSet` field identity label |
191 -| `<metric-name>` | `StateSet` special case: the flattened state name is carried under a synthetic label whose key is the base metric name |
191 +| Key / Pattern | Meaning |
192 +|-----------------|------------------------------------------------------------------------------------------------------------------------|
193 +| `le` | Histogram bucket bound label |
194 +| `quantile` | Summary quantile label |
195 +| `measure_field` | `MeasureSet` field identity label |
196 +| `<metric-name>` | `StateSet` special case: the flattened state name is carried under a synthetic label whose key is the base metric name |
197
198 Notes:
199
src/go/plugin/framework/chartengine/attempt.go new
+188
@@ -0,0 +1,188 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package chartengine
4 +
5 +import (
6 + "errors"
7 + "fmt"
8 + "sync"
9 +
10 + "github.com/netdata/netdata/go/plugins/pkg/metrix"
11 +)
12 +
13 +var (
14 + ErrOutstandingPlanAttempt = errors.New("chartengine: plan attempt already outstanding")
15 + ErrStalePlanAttempt = errors.New("chartengine: stale plan attempt")
16 + ErrFinishedPlanAttempt = errors.New("chartengine: plan attempt already finished")
17 +)
18 +
19 +type PlanAttempt struct {
20 + state *planAttemptState
21 +}
22 +
23 +type planAttemptState struct {
24 + mu sync.Mutex
25 +
26 + engine *Engine
27 + plan Plan
28 + materialized materializedState
29 + epoch uint64
30 + commitSeq uint64
31 + attemptID uint64
32 + reserved bool
33 + finished bool
34 +}
35 +
36 +func (a PlanAttempt) Plan() Plan {
37 + if a.state == nil {
38 + return Plan{}
39 + }
40 + return a.state.plan
41 +}
42 +
43 +func (a PlanAttempt) Commit() error {
44 + if a.state == nil {
45 + return nil
46 + }
47 +
48 + a.state.mu.Lock()
49 + if a.state.finished {
50 + a.state.mu.Unlock()
51 + return ErrFinishedPlanAttempt
52 + }
53 + a.state.finished = true
54 + reserved := a.state.reserved
55 + engine := a.state.engine
56 + materialized := a.state.materialized
57 + epoch := a.state.epoch
58 + commitSeq := a.state.commitSeq
59 + attemptID := a.state.attemptID
60 + a.state.mu.Unlock()
61 +
62 + if !reserved {
63 + return nil
64 + }
65 + if engine == nil {
66 + return fmt.Errorf("chartengine: nil engine on commit")
67 + }
68 + return engine.commitAttempt(materialized, epoch, commitSeq, attemptID)
69 +}
70 +
71 +func (a PlanAttempt) Abort() {
72 + if a.state == nil {
73 + return
74 + }
75 +
76 + a.state.mu.Lock()
77 + if a.state.finished {
78 + a.state.mu.Unlock()
79 + return
80 + }
81 + a.state.finished = true
82 + reserved := a.state.reserved
83 + engine := a.state.engine
84 + attemptID := a.state.attemptID
85 + a.state.mu.Unlock()
86 +
87 + if !reserved || engine == nil {
88 + return
89 + }
90 + engine.abortAttempt(attemptID)
91 +}
92 +
93 +func newPreparedAttempt(
94 + engine *Engine,
95 + plan Plan,
96 + materialized materializedState,
97 + epoch uint64,
98 + commitSeq uint64,
99 + attemptID uint64,
100 +) PlanAttempt {
101 + return PlanAttempt{
102 + state: &planAttemptState{
103 + engine: engine,
104 + plan: plan,
105 + materialized: materialized,
106 + epoch: epoch,
107 + commitSeq: commitSeq,
108 + attemptID: attemptID,
109 + reserved: true,
110 + },
111 + }
112 +}
113 +
114 +func newNoopAttempt(plan Plan) PlanAttempt {
115 + return PlanAttempt{
116 + state: &planAttemptState{
117 + plan: plan,
118 + },
119 + }
120 +}
121 +
122 +func (e *Engine) PreparePlan(reader metrix.Reader) (PlanAttempt, error) {
123 + plan, materialized, epoch, commitSeq, attemptID, reserved, err := e.preparePlan(reader)
124 + if err != nil {
125 + return PlanAttempt{}, err
126 + }
127 + if !reserved {
128 + return newNoopAttempt(plan), nil
129 + }
130 + return newPreparedAttempt(e, plan, materialized, epoch, commitSeq, attemptID), nil
131 +}
132 +
133 +func (e *Engine) nextAttemptIDLocked() uint64 {
134 + e.state.nextAttempt++
135 + if e.state.nextAttempt == 0 {
136 + e.state.nextAttempt++
137 + }
138 + return e.state.nextAttempt
139 +}
140 +
141 +func (e *Engine) commitAttempt(materialized materializedState, epoch, commitSeq, attemptID uint64) error {
142 + if e == nil {
143 + return fmt.Errorf("chartengine: nil engine")
144 + }
145 + e.mu.Lock()
146 + defer e.mu.Unlock()
147 +
148 + if e.state.outstanding != attemptID || e.state.outstanding == 0 {
149 + return ErrStalePlanAttempt
150 + }
151 + if e.state.engineEpoch != epoch || e.state.commitSeq != commitSeq {
152 + e.state.outstanding = 0
153 + return ErrStalePlanAttempt
154 + }
155 +
156 + e.state.materialized = materialized
157 + e.state.commitSeq++
158 + e.state.outstanding = 0
159 + return nil
160 +}
161 +
162 +func (e *Engine) abortAttempt(attemptID uint64) {
163 + if e == nil {
164 + return
165 + }
166 + e.mu.Lock()
167 + if e.state.outstanding == attemptID {
168 + e.state.outstanding = 0
169 + }
170 + e.mu.Unlock()
171 +}
172 +
173 +func prepareAndCommitPlan(engine *Engine, reader metrix.Reader) (Plan, error) {
174 + if engine == nil {
175 + return Plan{}, fmt.Errorf("chartengine: nil engine")
176 + }
177 + attempt, err := engine.PreparePlan(reader)
178 + if err != nil {
179 + return Plan{}, err
180 + }
181 + defer attempt.Abort()
182 +
183 + plan := attempt.Plan()
184 + if err := attempt.Commit(); err != nil {
185 + return Plan{}, err
186 + }
187 + return plan, nil
188 +}
src/go/plugin/framework/chartengine/engine.go
+16
@@ -77,6 +77,8 @@ func (e *Engine) Load(spec *charttpl.Spec, revision uint64) error {
77 // Template revision change resets routing/materialization internals.
78 e.state.routeCache = newRouteCache()
79 e.state.materialized = newMaterializedState()
80 + e.state.engineEpoch++
81 + e.state.outstanding = 0
82 e.mu.Unlock()
83 e.logInfof("chartengine program loaded revision=%d charts=%d metrics=%d", revision, len(compiled.Charts()), len(compiled.MetricNames()))
84 return nil
@@ -91,6 +93,20 @@ func (e *Engine) LoadYAML(data []byte, revision uint64) error {
93 return e.Load(spec, revision)
94 }
95
96 +// ResetMaterialized clears only materialized chart/dimension lifecycle state.
97 +//
98 +// It preserves the loaded program and other planner runtime state.
99 +func (e *Engine) ResetMaterialized() {
100 + if e == nil {
101 + return
102 + }
103 + e.mu.Lock()
104 + e.state.materialized = newMaterializedState()
105 + e.state.engineEpoch++
106 + e.state.outstanding = 0
107 + e.mu.Unlock()
108 +}
109 +
110 // loadYAMLFile reads chart-template YAML from file, compiles and publishes it.
111 func (e *Engine) loadYAMLFile(path string, revision uint64) error {
112 spec, err := charttpl.DecodeYAMLFile(path)
src/go/plugin/framework/chartengine/engine_test.go
+202
@@ -5,6 +5,7 @@ package chartengine
5 import (
6 "testing"
7
8 + "github.com/netdata/netdata/go/plugins/pkg/metrix"
9 "github.com/stretchr/testify/assert"
10 "github.com/stretchr/testify/require"
11 )
@@ -152,3 +153,204 @@ groups:
153 name: total
154 `
155 }
156 +
157 +func TestEngineResetMaterializedPreservesProgram(t *testing.T) {
158 + e, err := New()
159 + require.NoError(t, err)
160 + require.NoError(t, e.LoadYAML([]byte(validTemplateYAML()), 42))
161 +
162 + store := metrix.NewCollectorStore()
163 + cc := mustCycleController(t, store)
164 + cc.BeginCycle()
165 + store.Write().SnapshotMeter("").Gauge("mysql_queries_total").Observe(7)
166 + cc.CommitCycleSuccess()
167 +
168 + plan1, err := buildPlan(e, store.Read())
169 + require.NoError(t, err)
170 + require.NotNil(t, findCreateChartActionInEngineTests(plan1))
171 +
172 + p := e.program()
173 + require.NotNil(t, p)
174 + require.Equal(t, uint64(42), p.Revision())
175 +
176 + e.ResetMaterialized()
177 +
178 + p = e.program()
179 + require.NotNil(t, p)
180 + require.Equal(t, uint64(42), p.Revision())
181 +
182 + plan2, err := buildPlan(e, store.Read())
183 + require.NoError(t, err)
184 + require.NotNil(t, findCreateChartActionInEngineTests(plan2))
185 +}
186 +
187 +func TestEnginePreparePlanLifecycleScenarios(t *testing.T) {
188 + tests := map[string]struct {
189 + run func(t *testing.T)
190 + }{
191 + "outstanding attempt blocks prepare until abort": {
192 + run: func(t *testing.T) {
193 + e, err := New()
194 + require.NoError(t, err)
195 + require.NoError(t, e.LoadYAML([]byte(validTemplateYAML()), 1))
196 +
197 + store := metrix.NewCollectorStore()
198 + cc := mustCycleController(t, store)
199 + cc.BeginCycle()
200 + store.Write().SnapshotMeter("").Gauge("mysql_queries_total").Observe(7)
201 + cc.CommitCycleSuccess()
202 +
203 + attempt, err := e.PreparePlan(store.Read())
204 + require.NoError(t, err)
205 +
206 + _, err = e.PreparePlan(store.Read())
207 + require.ErrorIs(t, err, ErrOutstandingPlanAttempt)
208 +
209 + attempt.Abort()
210 + require.Empty(t, e.state.materialized.charts)
211 +
212 + cc.BeginCycle()
213 + store.Write().SnapshotMeter("").Gauge("mysql_queries_total").Observe(8)
214 + cc.CommitCycleSuccess()
215 +
216 + plan, err := buildPlan(e, store.Read())
217 + require.NoError(t, err)
218 + require.NotNil(t, findCreateChartActionInEngineTests(plan))
219 + },
220 + },
221 + "aborted attempt does not advance materialized lifecycle": {
222 + run: func(t *testing.T) {
223 + e, err := New()
224 + require.NoError(t, err)
225 + require.NoError(t, e.LoadYAML([]byte(validTemplateYAML()), 1))
226 +
227 + store := metrix.NewCollectorStore()
228 + cc := mustCycleController(t, store)
229 + cc.BeginCycle()
230 + store.Write().SnapshotMeter("").Gauge("mysql_queries_total").Observe(9)
231 + cc.CommitCycleSuccess()
232 +
233 + attempt, err := e.PreparePlan(store.Read())
234 + require.NoError(t, err)
235 + attempt.Abort()
236 + require.Empty(t, e.state.materialized.charts)
237 +
238 + cc.BeginCycle()
239 + store.Write().SnapshotMeter("").Gauge("mysql_queries_total").Observe(12)
240 + cc.CommitCycleSuccess()
241 +
242 + plan, err := buildPlan(e, store.Read())
243 + require.NoError(t, err)
244 + require.NotNil(t, findCreateChartActionInEngineTests(plan))
245 + },
246 + },
247 + "reset materialized makes prepared commit stale": {
248 + run: func(t *testing.T) {
249 + e, err := New()
250 + require.NoError(t, err)
251 + require.NoError(t, e.LoadYAML([]byte(validTemplateYAML()), 1))
252 +
253 + store := metrix.NewCollectorStore()
254 + cc := mustCycleController(t, store)
255 + cc.BeginCycle()
256 + store.Write().SnapshotMeter("").Gauge("mysql_queries_total").Observe(11)
257 + cc.CommitCycleSuccess()
258 +
259 + attempt, err := e.PreparePlan(store.Read())
260 + require.NoError(t, err)
261 +
262 + e.ResetMaterialized()
263 +
264 + require.ErrorIs(t, attempt.Commit(), ErrStalePlanAttempt)
265 + },
266 + },
267 + "repeated commit is rejected after successful commit": {
268 + run: func(t *testing.T) {
269 + e, err := New()
270 + require.NoError(t, err)
271 + require.NoError(t, e.LoadYAML([]byte(validTemplateYAML()), 1))
272 +
273 + store := metrix.NewCollectorStore()
274 + cc := mustCycleController(t, store)
275 + cc.BeginCycle()
276 + store.Write().SnapshotMeter("").Gauge("mysql_queries_total").Observe(13)
277 + cc.CommitCycleSuccess()
278 +
279 + attempt, err := e.PreparePlan(store.Read())
280 + require.NoError(t, err)
281 + require.NotNil(t, findCreateChartActionInEngineTests(attempt.Plan()))
282 +
283 + require.NoError(t, attempt.Commit())
284 + require.NotEmpty(t, e.state.materialized.charts)
285 + require.ErrorIs(t, attempt.Commit(), ErrFinishedPlanAttempt)
286 + },
287 + },
288 + }
289 +
290 + for name, tc := range tests {
291 + t.Run(name, tc.run)
292 + }
293 +}
294 +
295 +func TestEngineResetMaterializedKeepsRouteCacheWarm(t *testing.T) {
296 + e, err := New()
297 + require.NoError(t, err)
298 + require.NoError(t, e.LoadYAML([]byte(dynamicDimensionTemplateYAML()), 42))
299 +
300 + store := metrix.NewCollectorStore()
301 + cc := mustCycleController(t, store)
302 + vec := store.Write().StatefulMeter("component").Vec("id").Gauge("load")
303 +
304 + cc.BeginCycle()
305 + vec.WithLabelValues("a").Set(1)
306 + vec.WithLabelValues("b").Set(2)
307 + cc.CommitCycleSuccess()
308 +
309 + plan1, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
310 + require.NoError(t, err)
311 + require.NotNil(t, findCreateChartActionInEngineTests(plan1))
312 + stats1 := e.stats()
313 + require.Greater(t, stats1.RouteCacheMisses, uint64(0))
314 +
315 + plan2, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
316 + require.NoError(t, err)
317 + require.Nil(t, findCreateChartActionInEngineTests(plan2))
318 + stats2 := e.stats()
319 + require.Greater(t, stats2.RouteCacheHits, stats1.RouteCacheHits)
320 +
321 + e.ResetMaterialized()
322 +
323 + plan3, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
324 + require.NoError(t, err)
325 + require.NotNil(t, findCreateChartActionInEngineTests(plan3))
326 + stats3 := e.stats()
327 + require.Greater(t, stats3.RouteCacheHits, stats2.RouteCacheHits)
328 +}
329 +
330 +func findCreateChartActionInEngineTests(plan Plan) *CreateChartAction {
331 + for _, action := range plan.Actions {
332 + create, ok := action.(CreateChartAction)
333 + if ok {
334 + return &create
335 + }
336 + }
337 + return nil
338 +}
339 +
340 +func dynamicDimensionTemplateYAML() string {
341 + return `
342 +version: v1
343 +groups:
344 + - family: Runtime
345 + metrics:
346 + - component.load
347 + charts:
348 + - id: component_load
349 + title: Component Load
350 + context: netdata.go.plugin.component.component_load
351 + units: load
352 + dimensions:
353 + - selector: component.load
354 + name_from_label: id
355 +`
356 +}
src/go/plugin/framework/chartengine/helpers_test.go new
+9
@@ -0,0 +1,9 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package chartengine
4 +
5 +import "github.com/netdata/netdata/go/plugins/pkg/metrix"
6 +
7 +func buildPlan(engine *Engine, reader metrix.Reader) (Plan, error) {
8 + return prepareAndCommitPlan(engine, reader)
9 +}
src/go/plugin/framework/chartengine/lifecycle.go
+55
@@ -42,6 +42,61 @@ func newMaterializedState() materializedState {
42 }
43 }
44
45 +func (s materializedState) clone() materializedState {
46 + if len(s.charts) == 0 {
47 + return newMaterializedState()
48 + }
49 +
50 + out := materializedState{
51 + charts: make(map[string]*materializedChartState, len(s.charts)),
52 + }
53 + for chartID, chart := range s.charts {
54 + if chart == nil {
55 + continue
56 + }
57 + out.charts[chartID] = chart.clone()
58 + }
59 + return out
60 +}
61 +
62 +func (c *materializedChartState) clone() *materializedChartState {
63 + if c == nil {
64 + return nil
65 + }
66 +
67 + out := &materializedChartState{
68 + templateID: c.templateID,
69 + meta: c.meta,
70 + lifecycle: c.lifecycle,
71 + lastSeenSuccessSeq: c.lastSeenSuccessSeq,
72 + orderedDims: append([]string(nil), c.orderedDims...),
73 + orderedDimsDirty: c.orderedDimsDirty,
74 + }
75 + if len(c.dimensions) > 0 {
76 + out.dimensions = make(map[string]*materializedDimensionState, len(c.dimensions))
77 + for name, dim := range c.dimensions {
78 + if dim == nil {
79 + continue
80 + }
81 + cloned := *dim
82 + out.dimensions[name] = &cloned
83 + }
84 + } else {
85 + out.dimensions = make(map[string]*materializedDimensionState)
86 + }
87 + if len(c.scratchEntries) > 0 {
88 + out.scratchEntries = make(map[string]*dimBuildEntry, len(c.scratchEntries))
89 + for name, entry := range c.scratchEntries {
90 + if entry == nil {
91 + continue
92 + }
93 + cloned := *entry
94 + out.scratchEntries[name] = &cloned
95 + }
96 + }
97 + return out
98 +}
99 +
100 func (s *materializedState) ensureChart(
101 chartID string,
102 templateID string,
src/go/plugin/framework/chartengine/planner.go
+36 -38
@@ -106,6 +106,7 @@ type planBuildContext struct {
106 chartsByID map[string]*chartState
107 chartOwners map[string]string
108 dimCapHints map[string]int
109 + materialized *materializedState
110 materializedByID map[string]*materializedChartState
111
112 planRouteStats
@@ -115,18 +116,12 @@ type flattenedReadChecker interface {
116 FlattenedRead() bool
117 }
118
118 -// BuildPlan builds a minimal plan snapshot from the provided reader.
119 -//
120 -// Current scope:
121 -// - template routes with cache,
122 -// - optional unmatched-series autogen fallback,
123 -// - runtime-inferred dimension names from flattened metadata.
124 -func (e *Engine) BuildPlan(reader metrix.Reader) (Plan, error) {
119 +func (e *Engine) preparePlan(reader metrix.Reader) (Plan, materializedState, uint64, uint64, uint64, bool, error) {
120 if e == nil {
126 - return Plan{}, fmt.Errorf("chartengine: nil engine")
121 + return Plan{}, materializedState{}, 0, 0, 0, false, fmt.Errorf("chartengine: nil engine")
122 }
123 if reader == nil {
129 - return Plan{}, fmt.Errorf("chartengine: nil metrics reader")
124 + return Plan{}, materializedState{}, 0, 0, 0, false, fmt.Errorf("chartengine: nil metrics reader")
125 }
126 sample := planRuntimeSample{startedAt: time.Now()}
127 defer func() { e.observeBuildSample(sample) }()
@@ -136,16 +131,19 @@ func (e *Engine) BuildPlan(reader metrix.Reader) (Plan, error) {
131 InferredDimensions: make([]InferredDimension, 0),
132 }
133 collectMeta := reader.CollectMeta()
134 +
135 + e.mu.Lock()
136 + defer e.mu.Unlock()
137 + if e.state.outstanding != 0 {
138 + return Plan{}, materializedState{}, 0, 0, 0, false, ErrOutstandingPlanAttempt
139 + }
140 // Failed attempt must not trigger lifecycle transitions.
141 if collectMeta.LastAttemptStatus != metrix.CollectStatusSuccess {
142 sample.skippedFailed = true
143 e.logDebugf("chartengine build skipped: collect status=%d", collectMeta.LastAttemptStatus)
143 - return out, nil
144 + return out, materializedState{}, 0, 0, 0, false, nil
145 }
146
146 - e.mu.Lock()
147 - defer e.mu.Unlock()
148 -
147 obs := e.observeBuildSuccessSeq(collectMeta.LastSuccessSeq)
148 buildCycle := e.nextBuildCycle(collectMeta.LastSuccessSeq)
149 sample.buildSeqViolation = e.state.buildSeq.violating
@@ -168,19 +166,20 @@ func (e *Engine) BuildPlan(reader metrix.Reader) (Plan, error) {
166 }
167
168 phaseStartedAt := time.Now()
171 - ctx, err := e.preparePlanBuildContext(reader, &out, collectMeta, buildCycle)
169 + staged := e.state.materialized.clone()
170 + ctx, err := e.preparePlanBuildContext(reader, &out, collectMeta, buildCycle, &staged)
171 sample.phasePrepareSeconds = time.Since(phaseStartedAt).Seconds()
172 if err != nil {
173 sample.buildErr = true
174 e.logWarningf("chartengine build prepare failed: %v", err)
176 - return Plan{}, err
175 + return Plan{}, materializedState{}, 0, 0, 0, false, err
176 }
177 phaseStartedAt = time.Now()
178 if err := validateBuildReaderForInferredDimensions(ctx.index, reader); err != nil {
179 sample.phaseValidateSeconds = time.Since(phaseStartedAt).Seconds()
180 sample.buildErr = true
181 e.logWarningf("chartengine build reader validation failed: %v", err)
183 - return Plan{}, err
182 + return Plan{}, materializedState{}, 0, 0, 0, false, err
183 }
184 sample.phaseValidateSeconds = time.Since(phaseStartedAt).Seconds()
185 phaseStartedAt = time.Now()
@@ -188,7 +187,7 @@ func (e *Engine) BuildPlan(reader metrix.Reader) (Plan, error) {
187 sample.phaseScanSeconds = time.Since(phaseStartedAt).Seconds()
188 sample.buildErr = true
189 e.logWarningf("chartengine build scan failed: %v", err)
191 - return Plan{}, err
190 + return Plan{}, materializedState{}, 0, 0, 0, false, err
191 }
192 sample.phaseScanSeconds = time.Since(phaseStartedAt).Seconds()
193
@@ -202,7 +201,7 @@ func (e *Engine) BuildPlan(reader metrix.Reader) (Plan, error) {
201 sample.routeCacheFullDrop = retainStats.FullDrop
202
203 phaseStartedAt = time.Now()
205 - removeByCapDims, removeByCapCharts := enforceLifecycleCaps(ctx.collectMeta.LastSuccessSeq, ctx.chartsByID, &e.state.materialized)
204 + removeByCapDims, removeByCapCharts := enforceLifecycleCaps(ctx.collectMeta.LastSuccessSeq, ctx.chartsByID, ctx.materialized)
205 sample.phaseLifecycleCapsSec = time.Since(phaseStartedAt).Seconds()
206 sample.lifecycleRemovedDimensionByCap = len(removeByCapDims)
207 sample.lifecycleRemovedChartByCap = len(removeByCapCharts)
@@ -217,11 +216,11 @@ func (e *Engine) BuildPlan(reader metrix.Reader) (Plan, error) {
216 sample.phaseMaterializeSeconds = time.Since(phaseStartedAt).Seconds()
217 sample.buildErr = true
218 e.logWarningf("chartengine build materialization failed: %v", err)
220 - return Plan{}, err
219 + return Plan{}, materializedState{}, 0, 0, 0, false, err
220 }
221 sample.phaseMaterializeSeconds = time.Since(phaseStartedAt).Seconds()
222 phaseStartedAt = time.Now()
224 - removeDims, removeCharts := collectExpiryRemovals(ctx.collectMeta.LastSuccessSeq, &e.state.materialized)
223 + removeDims, removeCharts := collectExpiryRemovals(ctx.collectMeta.LastSuccessSeq, ctx.materialized)
224 sample.phaseExpirySeconds = time.Since(phaseStartedAt).Seconds()
225 sample.lifecycleRemovedDimensionByExpiry = len(removeDims)
226 sample.lifecycleRemovedChartByExpiry = len(removeCharts)
@@ -249,7 +248,9 @@ func (e *Engine) BuildPlan(reader metrix.Reader) (Plan, error) {
248 e.state.hints.chartsByID = len(ctx.chartsByID)
249 e.state.hints.seenInfer = len(ctx.seenInfer)
250
252 - return out, nil
251 + attemptID := e.nextAttemptIDLocked()
252 + e.state.outstanding = attemptID
253 + return out, staged, e.state.engineEpoch, e.state.commitSeq, attemptID, true, nil
254 }
255
256 func validateBuildReaderForInferredDimensions(index matchIndex, reader metrix.Reader) error {
@@ -293,6 +294,7 @@ func (e *Engine) preparePlanBuildContext(
294 out *Plan,
295 collectMeta metrix.CollectMeta,
296 buildCycle uint64,
297 + materialized *materializedState,
298 ) (*planBuildContext, error) {
299 prog := e.state.program
300 if prog == nil {
@@ -303,25 +305,28 @@ func (e *Engine) preparePlanBuildContext(
305 cache = newRouteCache()
306 e.state.routeCache = cache
307 }
306 - if e.state.materialized.charts == nil {
307 - e.state.materialized = newMaterializedState()
308 + if materialized == nil {
309 + return nil, fmt.Errorf("chartengine: nil materialized state")
310 + }
311 + if materialized.charts == nil {
312 + *materialized = newMaterializedState()
313 }
314 index := e.state.matchIndex
315 if index.chartsByID == nil {
316 index = buildMatchIndex(prog.Charts())
317 e.state.matchIndex = index
318 }
314 - chartOwners := make(map[string]string, len(e.state.materialized.charts))
315 - dimCapHints := make(map[string]int, len(e.state.materialized.charts))
316 - for chartID, matChart := range e.state.materialized.charts {
319 + chartOwners := make(map[string]string, len(materialized.charts))
320 + dimCapHints := make(map[string]int, len(materialized.charts))
321 + for chartID, matChart := range materialized.charts {
322 chartOwners[chartID] = matChart.templateID
323 if n := len(matChart.dimensions); n > 0 {
324 dimCapHints[chartID] = n
325 }
326 }
327 chartsCap := e.state.hints.chartsByID
323 - if chartsCap < len(e.state.materialized.charts) {
324 - chartsCap = len(e.state.materialized.charts)
328 + if chartsCap < len(materialized.charts) {
329 + chartsCap = len(materialized.charts)
330 }
331 seenInferCap := e.state.hints.seenInfer
332 return &planBuildContext{
@@ -337,7 +342,8 @@ func (e *Engine) preparePlanBuildContext(
342 chartsByID: make(map[string]*chartState, chartsCap),
343 chartOwners: chartOwners,
344 dimCapHints: dimCapHints,
340 - materializedByID: e.state.materialized.charts,
345 + materialized: materialized,
346 + materializedByID: materialized.charts,
347 }, nil
348 }
349
@@ -550,7 +556,7 @@ func (e *Engine) materializePlanCharts(ctx *planBuildContext) error {
556
557 for _, chartID := range chartIDs {
558 cs := ctx.chartsByID[chartID]
553 - matChart, chartCreated := e.state.materialized.ensureChart(cs.chartID, cs.templateID, cs.meta, cs.lifecycle)
559 + matChart, chartCreated := ctx.materialized.ensureChart(cs.chartID, cs.templateID, cs.meta, cs.lifecycle)
560 if chartCreated {
561 chartLabels := map[string]string(nil)
562 if cs.labels != nil {
@@ -660,14 +666,6 @@ func sortInferredDimensions(in []InferredDimension) {
666 })
667 }
668
663 -// buildPlan is a package-level convenience wrapper around Engine.BuildPlan.
664 -func buildPlan(engine *Engine, reader metrix.Reader) (Plan, error) {
665 - if engine == nil {
666 - return Plan{}, fmt.Errorf("chartengine: nil engine")
667 - }
668 - return engine.BuildPlan(reader)
669 -}
670 -
669 func isAutogenTemplateID(templateID string) bool {
670 return strings.HasPrefix(templateID, autogenTemplatePrefix)
671 }
src/go/plugin/framework/chartengine/planner_bench_test.go
+1 -1
@@ -45,7 +45,7 @@ func BenchmarkBuildPlanBySeriesCardinality(b *testing.B) {
45 b.ReportAllocs()
46 b.ResetTimer()
47 for i := 0; i < b.N; i++ {
48 - if _, err := engine.BuildPlan(reader); err != nil {
48 + if _, err := buildPlan(engine, reader); err != nil {
49 b.Fatalf("build plan: %v", err)
50 }
51 }
src/go/plugin/framework/chartengine/planner_test.go
+56 -58
@@ -184,7 +184,7 @@ groups:
184 store := metrix.NewCollectorStore()
185 tc.setup(t, store)
186
187 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
187 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
188 require.NoError(t, err)
189
190 got := make([]string, 0, len(plan.InferredDimensions))
@@ -270,11 +270,11 @@ groups:
270 ss.Enable("ok")
271 cc.CommitCycleSuccess()
272
273 - _, err = e.BuildPlan(store.Read())
273 + _, err = buildPlan(e, store.Read())
274 require.Error(t, err)
275 assert.ErrorContains(t, err, "Read(metrix.ReadFlatten())")
276
277 - _, err = e.BuildPlan(store.Read(metrix.ReadFlatten()))
277 + _, err = buildPlan(e, store.Read(metrix.ReadFlatten()))
278 require.NoError(t, err)
279 }
280
@@ -306,7 +306,7 @@ groups:
306 c.ObserveTotal(10)
307 cc.CommitCycleSuccess()
308
309 - plan1, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
309 + plan1, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
310 require.NoError(t, err)
311 assert.Equal(t, []ActionKind{ActionCreateChart, ActionCreateDimension, ActionUpdateChart}, actionKinds(plan1.Actions))
312 stats1 := e.stats()
@@ -319,7 +319,7 @@ groups:
319 c.ObserveTotal(20)
320 cc.CommitCycleSuccess()
321
322 - plan2, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
322 + plan2, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
323 require.NoError(t, err)
324 assert.Equal(t, []ActionKind{ActionUpdateChart}, actionKinds(plan2.Actions))
325 stats2 := e.stats()
@@ -367,7 +367,7 @@ groups:
367 modeMetric.Observe(1, modeOK)
368 cc.CommitCycleSuccess()
369
370 - plan1, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
370 + plan1, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
371 require.NoError(t, err)
372 assert.Equal(t, []ActionKind{ActionCreateChart, ActionCreateDimension, ActionCreateDimension, ActionUpdateChart}, actionKinds(plan1.Actions))
373
@@ -375,7 +375,7 @@ groups:
375 total.Observe(101)
376 cc.CommitCycleSuccess()
377
378 - plan2, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
378 + plan2, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
379 require.NoError(t, err)
380 assert.Equal(t, []ActionKind{ActionUpdateChart, ActionRemoveDimension}, actionKinds(plan2.Actions))
381 removeDim := findRemoveDimensionAction(plan2)
@@ -413,14 +413,14 @@ groups:
413 c.ObserveTotal(10)
414 cc.CommitCycleSuccess()
415
416 - plan1, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
416 + plan1, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
417 require.NoError(t, err)
418 assert.Equal(t, []ActionKind{ActionCreateChart, ActionCreateDimension, ActionUpdateChart}, actionKinds(plan1.Actions))
419
420 cc.BeginCycle()
421 cc.CommitCycleSuccess()
422
423 - plan2, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
423 + plan2, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
424 require.NoError(t, err)
425 assert.Equal(t, []ActionKind{ActionRemoveChart}, actionKinds(plan2.Actions))
426 }
@@ -455,21 +455,21 @@ groups:
455 c.ObserveTotal(10)
456 cc.CommitCycleSuccess()
457
458 - plan1, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
458 + plan1, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
459 require.NoError(t, err)
460 assert.Equal(t, []ActionKind{ActionCreateChart, ActionCreateDimension, ActionUpdateChart}, actionKinds(plan1.Actions))
461
462 cc.BeginCycle()
463 cc.AbortCycle()
464
465 - plan2, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
465 + plan2, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
466 require.NoError(t, err)
467 assert.Empty(t, plan2.Actions)
468
469 cc.BeginCycle()
470 cc.CommitCycleSuccess()
471
472 - plan3, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
472 + plan3, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
473 require.NoError(t, err)
474 assert.Equal(t, []ActionKind{ActionRemoveChart}, actionKinds(plan3.Actions))
475 }
@@ -510,7 +510,7 @@ groups:
510 rx.ObserveTotal(20, eth0)
511 cc.CommitCycleSuccess()
512
513 - plan1, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
513 + plan1, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
514 require.NoError(t, err)
515 assert.Equal(t, []ActionKind{
516 ActionCreateChart, ActionCreateDimension, ActionUpdateChart,
@@ -539,7 +539,7 @@ groups:
539 rx.ObserveTotal(11, eth1)
540 cc.CommitCycleSuccess()
541
542 - plan2, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
542 + plan2, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
543 require.NoError(t, err)
544 assert.Equal(t, []ActionKind{ActionUpdateChart, ActionUpdateChart}, actionKinds(plan2.Actions))
545 }
@@ -581,7 +581,7 @@ groups:
581 rx.ObserveTotal(10, eth0)
582 cc.CommitCycleSuccess()
583
584 - plan1, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
584 + plan1, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
585 require.NoError(t, err)
586 assert.Equal(t, []ActionKind{ActionCreateChart, ActionCreateDimension, ActionUpdateChart}, actionKinds(plan1.Actions))
587
@@ -590,7 +590,7 @@ groups:
590 rx.ObserveTotal(20, eth1)
591 cc.CommitCycleSuccess()
592
593 - plan2, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
593 + plan2, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
594 require.NoError(t, err)
595 // eth0 exists and is seen, so eth1 is dropped under max_instances=1.
596 assert.Equal(t, []ActionKind{ActionUpdateChart}, actionKinds(plan2.Actions))
@@ -602,7 +602,7 @@ groups:
602 rx.ObserveTotal(21, eth1)
603 cc.CommitCycleSuccess()
604
605 - plan3, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
605 + plan3, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
606 require.NoError(t, err)
607 assert.Equal(t, []ActionKind{
608 ActionRemoveChart,
@@ -650,7 +650,7 @@ groups:
650 g.Observe(1, modeB)
651 cc.CommitCycleSuccess()
652
653 - plan1, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
653 + plan1, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
654 require.NoError(t, err)
655 assert.Equal(t, []ActionKind{ActionCreateChart, ActionCreateDimension, ActionCreateDimension, ActionUpdateChart}, actionKinds(plan1.Actions))
656
@@ -660,7 +660,7 @@ groups:
660 g.Observe(1, modeC)
661 cc.CommitCycleSuccess()
662
663 - plan2, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
663 + plan2, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
664 require.NoError(t, err)
665 // a,b seen; c is dropped under max_dims=2.
666 assert.Equal(t, []ActionKind{ActionUpdateChart}, actionKinds(plan2.Actions))
@@ -673,7 +673,7 @@ groups:
673 g.Observe(1, modeC)
674 cc.CommitCycleSuccess()
675
676 - plan3, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
676 + plan3, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
677 require.NoError(t, err)
678 assert.Equal(t, []ActionKind{
679 ActionRemoveDimension,
@@ -727,7 +727,7 @@ groups:
727 m.ObserveTotal(20, out)
728 cc.CommitCycleSuccess()
729
730 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
730 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
731 require.NoError(t, err)
732
733 var create *CreateChartAction
@@ -772,7 +772,7 @@ groups:
772 unmatched.ObserveTotal(10)
773 cc.CommitCycleSuccess()
774
775 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
775 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
776 require.NoError(t, err)
777 assert.Empty(t, plan.Actions)
778 }
@@ -815,7 +815,7 @@ groups:
815 unmatched.ObserveTotal(20, methodPOST)
816 cc.CommitCycleSuccess()
817
818 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
818 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
819 require.NoError(t, err)
820
821 assert.Equal(t, []ActionKind{ActionCreateChart, ActionCreateDimension, ActionUpdateChart}, actionKinds(plan.Actions))
@@ -859,7 +859,7 @@ groups:
859 unmatched.ObserveTotal(20, methodPOST)
860 cc.CommitCycleSuccess()
861
862 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
862 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
863 require.NoError(t, err)
864
865 assert.Equal(t, []ActionKind{ActionCreateChart, ActionCreateDimension, ActionUpdateChart}, actionKinds(plan.Actions))
@@ -903,7 +903,7 @@ groups:
903 unmatched.ObserveTotal(20, methodPOST)
904 cc.CommitCycleSuccess()
905
906 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
906 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
907 require.NoError(t, err)
908
909 assert.Equal(t, []ActionKind{ActionCreateChart, ActionCreateDimension, ActionUpdateChart}, actionKinds(plan.Actions))
@@ -941,7 +941,7 @@ groups:
941 unmatched.ObserveTotal(20, methodPOST)
942 cc.CommitCycleSuccess()
943
944 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
944 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
945 require.NoError(t, err)
946
947 assert.Equal(t, []ActionKind{ActionCreateChart, ActionCreateDimension, ActionUpdateChart}, actionKinds(plan.Actions))
@@ -980,7 +980,7 @@ groups:
980 unmatched.ObserveTotal(10, methodGET)
981 cc.CommitCycleSuccess()
982
983 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
983 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
984 require.NoError(t, err)
985
986 assert.Equal(t, []ActionKind{ActionCreateChart, ActionCreateDimension, ActionUpdateChart}, actionKinds(plan.Actions))
@@ -1031,7 +1031,7 @@ groups:
1031 unmatched.ObserveTotal(10)
1032 cc.CommitCycleSuccess()
1033
1034 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
1034 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
1035 require.NoError(t, err)
1036
1037 create := findCreateChartAction(plan)
@@ -1082,7 +1082,7 @@ groups:
1082 })
1083 cc.CommitCycleSuccess()
1084
1085 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
1085 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
1086 require.NoError(t, err)
1087
1088 buckets := findCreateChartActionByID(plan, "svc.request_duration_ms")
@@ -1129,7 +1129,7 @@ groups:
1129 unmatched.Observe(10.5)
1130 cc.CommitCycleSuccess()
1131
1132 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
1132 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
1133 require.NoError(t, err)
1134
1135 var created *CreateDimensionAction
@@ -1186,7 +1186,7 @@ groups:
1186 })
1187 cc.CommitCycleSuccess()
1188
1189 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
1189 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
1190 require.NoError(t, err)
1191
1192 sum := findCreateChartActionByID(plan, "svc.query_duration_ms_sum")
@@ -1227,7 +1227,7 @@ groups:
1227 m.ObserveTotal(10, methodGET)
1228 cc.CommitCycleSuccess()
1229
1230 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
1230 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
1231 require.NoError(t, err)
1232 assert.Equal(t, []ActionKind{ActionCreateChart, ActionCreateDimension, ActionUpdateChart}, actionKinds(plan.Actions))
1233 create := findCreateChartAction(plan)
@@ -1288,7 +1288,7 @@ groups:
1288 metric.ObserveTotal(10, ls)
1289 cc.CommitCycleSuccess()
1290
1291 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
1291 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
1292 require.NoError(t, err)
1293 assert.Empty(t, plan.Actions)
1294 }
@@ -1330,7 +1330,7 @@ groups:
1330 }, method)
1331 cc.CommitCycleSuccess()
1332
1333 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
1333 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
1334 require.NoError(t, err)
1335
1336 var bucketChart *CreateChartAction
@@ -1392,7 +1392,7 @@ groups:
1392 g.Observe(7, queueMain)
1393 cc.CommitCycleSuccess()
1394
1395 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
1395 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
1396 require.NoError(t, err)
1397
1398 assert.Equal(t, []ActionKind{ActionCreateChart, ActionCreateDimension, ActionUpdateChart}, actionKinds(plan.Actions))
@@ -1442,7 +1442,7 @@ groups:
1442 ss.Enable("operational")
1443 cc.CommitCycleSuccess()
1444
1445 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
1445 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
1446 require.NoError(t, err)
1447
1448 assert.Equal(t, []ActionKind{
@@ -1505,7 +1505,7 @@ groups:
1505 ss.Enable("operational")
1506 cc.CommitCycleSuccess()
1507
1508 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
1508 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
1509 require.NoError(t, err)
1510
1511 create := findCreateChartAction(plan)
@@ -1553,7 +1553,7 @@ groups:
1553 ms.ObservePoint(metrix.MeasureSetPoint{Values: []metrix.SampleValue{1.5, 0.5}})
1554 cc.CommitCycleSuccess()
1555
1556 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
1556 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
1557 require.NoError(t, err)
1558
1559 assert.Equal(t, []ActionKind{
@@ -1635,7 +1635,7 @@ groups:
1635 ms.ObserveTotalPoint(metrix.MeasureSetPoint{Values: []metrix.SampleValue{10, 2}})
1636 cc.CommitCycleSuccess()
1637
1638 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
1638 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
1639 require.NoError(t, err)
1640
1641 assert.Equal(t, []ActionKind{
@@ -1713,7 +1713,7 @@ groups:
1713 fooTotal.ObserveTotal(7, methodGET)
1714 cc.CommitCycleSuccess()
1715
1716 - plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
1716 + plan, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
1717 require.NoError(t, err)
1718
1719 assert.Equal(t, []ActionKind{ActionCreateChart, ActionCreateDimension, ActionUpdateChart}, actionKinds(plan.Actions))
@@ -1760,14 +1760,14 @@ groups:
1760 c.ObserveTotal(10)
1761 cc.CommitCycleSuccess()
1762
1763 - plan1, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
1763 + plan1, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
1764 require.NoError(t, err)
1765 assert.Equal(t, []ActionKind{ActionCreateChart, ActionCreateDimension, ActionUpdateChart}, actionKinds(plan1.Actions))
1766
1767 cc.BeginCycle()
1768 cc.CommitCycleSuccess()
1769
1770 - plan2, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
1770 + plan2, err := buildPlan(e, store.Read(metrix.ReadFlatten()))
1771 require.NoError(t, err)
1772 assert.Equal(t, []ActionKind{ActionRemoveChart}, actionKinds(plan2.Actions))
1773 }
@@ -1814,7 +1814,7 @@ groups:
1814 b.Observe(3, total)
1815 cc.CommitCycleSuccess()
1816
1817 - plan, err := e.BuildPlan(store.Read())
1817 + plan, err := buildPlan(e, store.Read())
1818 require.NoError(t, err)
1819
1820 assert.Equal(t, []ActionKind{
@@ -1881,7 +1881,7 @@ groups:
1881 mode.Observe(2, warnSet)
1882 cc.CommitCycleSuccess()
1883
1884 - plan1, err := e.BuildPlan(store.Read())
1884 + plan1, err := buildPlan(e, store.Read())
1885 require.NoError(t, err)
1886 require.NotNil(t, findUpdateAction(plan1))
1887
@@ -1889,14 +1889,13 @@ groups:
1889 require.NotNil(t, matChart)
1890 require.Contains(t, matChart.scratchEntries, "ok")
1891 require.Contains(t, matChart.scratchEntries, "warn")
1892 - okEntryPtr := matChart.scratchEntries["ok"]
1893 - require.NotNil(t, okEntryPtr)
1892 + require.NotNil(t, matChart.scratchEntries["ok"])
1893
1894 cc.BeginCycle()
1895 mode.Observe(3, okSet)
1896 cc.CommitCycleSuccess()
1897
1899 - plan2, err := e.BuildPlan(store.Read())
1898 + plan2, err := buildPlan(e, store.Read())
1899 require.NoError(t, err)
1900
1901 update2 := findUpdateAction(plan2)
@@ -1916,13 +1915,12 @@ groups:
1915 assert.NotContains(t, matChart.dimensions, "warn")
1916 require.Contains(t, matChart.scratchEntries, "warn")
1917 require.Contains(t, matChart.scratchEntries, "ok")
1919 - assert.Equal(t, okEntryPtr, matChart.scratchEntries["ok"])
1918
1919 cc.BeginCycle()
1920 mode.Observe(4, okSet)
1921 cc.CommitCycleSuccess()
1922
1925 - plan3, err := e.BuildPlan(store.Read())
1923 + plan3, err := buildPlan(e, store.Read())
1924 require.NoError(t, err)
1925 update3 := findUpdateAction(plan3)
1926 require.NotNil(t, update3)
@@ -1967,11 +1965,11 @@ groups:
1965 g.Observe(5)
1966 cc.CommitCycleSuccess()
1967
1970 - plan1, err := e.BuildPlan(store.Read())
1968 + plan1, err := buildPlan(e, store.Read())
1969 require.NoError(t, err)
1970 require.NotNil(t, findUpdateAction(plan1))
1971
1974 - plan2, err := e.BuildPlan(store.Read())
1972 + plan2, err := buildPlan(e, store.Read())
1973 require.NoError(t, err)
1974 assert.Empty(t, plan2.Actions)
1975 },
@@ -2002,7 +2000,7 @@ groups:
2000 vec.WithLabelValues("warn").Set(2)
2001
2002 reader := store.Read(metrix.ReadRaw(), metrix.ReadFlatten())
2005 - plan1, err := e.BuildPlan(reader)
2003 + plan1, err := buildPlan(e, reader)
2004 require.NoError(t, err)
2005 require.NotNil(t, findUpdateAction(plan1))
2006
@@ -2010,10 +2008,9 @@ groups:
2008 require.NotNil(t, matChart)
2009 require.Contains(t, matChart.scratchEntries, "ok")
2010 require.Contains(t, matChart.scratchEntries, "warn")
2013 - okEntry := matChart.scratchEntries["ok"]
2014 - require.NotNil(t, okEntry)
2011 + require.NotNil(t, matChart.scratchEntries["ok"])
2012
2016 - plan2, err := e.BuildPlan(reader)
2013 + plan2, err := buildPlan(e, reader)
2014 require.NoError(t, err)
2015 assert.Equal(t, []ActionKind{ActionUpdateChart}, actionKinds(plan2.Actions))
2016 require.NotNil(t, findUpdateAction(plan2))
@@ -2029,7 +2026,6 @@ groups:
2026 require.NotNil(t, matChart)
2027 require.Contains(t, matChart.scratchEntries, "ok")
2028 require.Contains(t, matChart.scratchEntries, "warn")
2032 - assert.Equal(t, okEntry, matChart.scratchEntries["ok"])
2029 },
2030 },
2031 }
@@ -2080,7 +2076,8 @@ groups:
2076 }
2077 reader := store.Read()
2078 meta := reader.CollectMeta()
2083 - ctx, err := e.preparePlanBuildContext(reader, &out, meta, meta.LastSuccessSeq)
2079 + materialized := e.state.materialized.clone()
2080 + ctx, err := e.preparePlanBuildContext(reader, &out, meta, meta.LastSuccessSeq, &materialized)
2081 require.NoError(t, err)
2082 require.NoError(t, e.scanPlanSeries(ctx))
2083
@@ -2130,7 +2127,8 @@ groups:
2127 }
2128 reader := store.Read()
2129 meta := reader.CollectMeta()
2133 - ctx, err := e.preparePlanBuildContext(reader, &out, meta, meta.LastSuccessSeq)
2130 + materialized := e.state.materialized.clone()
2131 + ctx, err := e.preparePlanBuildContext(reader, &out, meta, meta.LastSuccessSeq, &materialized)
2132 require.NoError(t, err)
2133 require.NoError(t, e.scanPlanSeries(ctx))
2134 require.NoError(t, e.materializePlanCharts(ctx))
src/go/plugin/framework/chartengine/runtime_metrics_test.go
+10 -10
@@ -29,13 +29,13 @@ func TestEngineRuntimeObservabilityScenarios(t *testing.T) {
29 cc.BeginCycle()
30 c.ObserveTotal(10)
31 cc.CommitCycleSuccess()
32 - _, err = e.BuildPlan(store.Read(metrix.ReadFlatten()))
32 + _, err = buildPlan(e, store.Read(metrix.ReadFlatten()))
33 require.NoError(t, err)
34
35 cc.BeginCycle()
36 c.ObserveTotal(20)
37 cc.CommitCycleSuccess()
38 - _, err = e.BuildPlan(store.Read(metrix.ReadFlatten()))
38 + _, err = buildPlan(e, store.Read(metrix.ReadFlatten()))
39 require.NoError(t, err)
40
41 rs := e.RuntimeStore()
@@ -87,7 +87,7 @@ func TestEngineRuntimeObservabilityScenarios(t *testing.T) {
87 cc.BeginCycle()
88 c.ObserveTotal(10)
89 cc.CommitCycleSuccess()
90 - _, err = e.BuildPlan(store.Read(metrix.ReadFlatten()))
90 + _, err = buildPlan(e, store.Read(metrix.ReadFlatten()))
91 require.NoError(t, err)
92
93 before := e.RuntimeStore().Read(metrix.ReadRaw())
@@ -97,7 +97,7 @@ func TestEngineRuntimeObservabilityScenarios(t *testing.T) {
97
98 cc.BeginCycle()
99 cc.AbortCycle()
100 - _, err = e.BuildPlan(store.Read(metrix.ReadFlatten()))
100 + _, err = buildPlan(e, store.Read(metrix.ReadFlatten()))
101 require.NoError(t, err)
102
103 after := e.RuntimeStore().Read(metrix.ReadRaw())
@@ -125,7 +125,7 @@ func TestEngineRuntimeObservabilityScenarios(t *testing.T) {
125 cc.BeginCycle()
126 c.ObserveTotal(10)
127 cc.CommitCycleSuccess()
128 - _, err = e.BuildPlan(store.Read(metrix.ReadFlatten()))
128 + _, err = buildPlan(e, store.Read(metrix.ReadFlatten()))
129 require.NoError(t, err)
130
131 r := e.RuntimeStore().Read(metrix.ReadRaw())
@@ -149,13 +149,13 @@ func TestEngineRuntimeObservabilityScenarios(t *testing.T) {
149 total.Observe(100)
150 modeMetric.Observe(1, modeOK)
151 cc.CommitCycleSuccess()
152 - _, err = e.BuildPlan(store.Read(metrix.ReadFlatten()))
152 + _, err = buildPlan(e, store.Read(metrix.ReadFlatten()))
153 require.NoError(t, err)
154
155 cc.BeginCycle()
156 total.Observe(101)
157 cc.CommitCycleSuccess()
158 - _, err = e.BuildPlan(store.Read(metrix.ReadFlatten()))
158 + _, err = buildPlan(e, store.Read(metrix.ReadFlatten()))
159 require.NoError(t, err)
160
161 r := e.RuntimeStore().Read(metrix.ReadRaw())
@@ -186,7 +186,7 @@ func TestEngineRuntimeObservabilityScenarios(t *testing.T) {
186 require.NoError(t, err)
187 require.NoError(t, observer.LoadYAML([]byte(runtimeComponentTemplateYAML()), 1))
188
189 - plan, err := observer.BuildPlan(rs.Read(metrix.ReadFlatten()))
189 + plan, err := buildPlan(observer, rs.Read(metrix.ReadFlatten()))
190 require.NoError(t, err)
191 assert.Equal(t, []ActionKind{ActionCreateChart, ActionCreateDimension, ActionUpdateChart}, actionKinds(plan.Actions))
192
@@ -211,7 +211,7 @@ func TestEngineRuntimeObservabilityScenarios(t *testing.T) {
211 cc.BeginCycle()
212 c.ObserveTotal(10)
213 cc.CommitCycleSuccess()
214 - _, err = producer.BuildPlan(store.Read(metrix.ReadFlatten()))
214 + _, err = buildPlan(producer, store.Read(metrix.ReadFlatten()))
215 require.NoError(t, err)
216
217 observer, err := New(
@@ -223,7 +223,7 @@ func TestEngineRuntimeObservabilityScenarios(t *testing.T) {
223 require.NoError(t, err)
224 require.NoError(t, observer.LoadYAML([]byte(runtimeDummyTemplateYAML()), 1))
225
226 - plan, err := observer.BuildPlan(producer.RuntimeStore().Read(metrix.ReadFlatten()))
226 + plan, err := buildPlan(observer, producer.RuntimeStore().Read(metrix.ReadFlatten()))
227 require.NoError(t, err)
228 create := findCreateChartByTitle(plan.Actions, "Successful BuildPlan calls")
229 require.NotNil(t, create)
src/go/plugin/framework/chartengine/state.go
+4
@@ -14,6 +14,10 @@ type engineState struct {
14 matchIndex matchIndex
15 routeCache *routeCache
16 materialized materializedState
17 + engineEpoch uint64
18 + commitSeq uint64
19 + nextAttempt uint64
20 + outstanding uint64
21 hints plannerSizingHints
22 buildSeq buildSeqState
23 // plannerBuildSeq is runtime-mode build-cycle sequence used only by
src/go/plugin/framework/jobruntime/job_v1.go
+10 -11
@@ -16,6 +16,7 @@ import (
16
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/chartemit"
20 "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
21 "github.com/netdata/netdata/go/plugins/plugin/framework/metricsaudit"
22 "github.com/netdata/netdata/go/plugins/plugin/framework/tickstate"
@@ -577,21 +578,19 @@ func (j *Job) processMetrics(mx collectedMetrics, startTime time.Time, sinceLast
578 }
579
580 func (j *Job) sendVnodeHostInfo() {
580 - if j.vnode.Labels == nil {
581 - j.vnode.Labels = make(map[string]string)
582 - }
583 - if _, ok := j.vnode.Labels["_hostname"]; !ok {
584 - j.vnode.Labels["_hostname"] = j.vnode.Hostname
585 - }
586 - for k, v := range j.vnode.Labels {
587 - j.vnode.Labels[k] = lblValueReplacer.Replace(v)
588 - }
589 -
590 - j.api.HOSTINFO(netdataapi.HostInfo{
581 + info, err := chartemit.PrepareHostInfo(netdataapi.HostInfo{
582 GUID: j.vnode.GUID,
583 Hostname: j.vnode.Hostname,
584 Labels: j.vnode.Labels,
585 })
586 + if err != nil {
587 + j.Warningf("prepare vnode host info failed: %v", err)
588 + return
589 + }
590 +
591 + j.vnode.Hostname = info.Hostname
592 + j.vnode.Labels = info.Labels
593 + j.api.HOSTINFO(info)
594 }
595
596 func (j *Job) createChart(chart *collectorapi.Chart) {
src/go/plugin/framework/jobruntime/job_v2.go
+105 -16
@@ -117,6 +117,8 @@ type JobV2 struct {
117 vnode vnodes.VirtualNode
118 updVnode chan *vnodes.VirtualNode
119
120 + hostState jobV2HostState
121 +
122 ctxMu sync.RWMutex
123 runCtx context.Context
124 cancelRun context.CancelFunc
@@ -135,6 +137,12 @@ type JobV2 struct {
137 skipTracker tickstate.SkipTracker
138 }
139
140 +type jobV2PreparedEmission struct {
141 + attempt chartengine.PlanAttempt
142 + plan chartengine.Plan
143 + decision jobV2EmissionDecision
144 +}
145 +
146 func (j *JobV2) FullName() string { return j.fullName }
147 func (j *JobV2) ModuleName() string { return j.moduleName }
148 func (j *JobV2) Name() string { return j.name }
@@ -171,10 +179,38 @@ func (j *JobV2) UpdateVnode(vnode *vnodes.VirtualNode) {
179 j.updVnode <- vnode
180 }
181 func (j *JobV2) Cleanup() {
182 + j.buf.Reset()
183 + snapshot := j.hostState.captureCleanupSnapshot(j.currentVnode())
184 j.unregisterRuntimeComponent()
185 if j.module != nil {
186 j.module.Cleanup(context.Background())
187 }
188 + if !collectorapi.ShouldObsoleteCharts() {
189 + return
190 + }
191 + if snapshot.staleVnodeSuppressed || len(snapshot.charts) == 0 {
192 + return
193 + }
194 +
195 + env := chartemit.EmitEnv{
196 + TypeID: j.fullName,
197 + UpdateEvery: j.updateEvery,
198 + Plugin: j.pluginName,
199 + Module: j.moduleName,
200 + JobName: j.name,
201 + JobLabels: j.labels,
202 + }
203 + if snapshot.host.isVnode() {
204 + env.HostScope = &chartemit.HostScope{GUID: snapshot.host.guid}
205 + }
206 + if err := chartemit.ApplyPlan(j.api, buildJobV2CleanupPlan(snapshot.charts), env); err != nil {
207 + j.Warningf("cleanup apply plan failed: %v", err)
208 + j.buf.Reset()
209 + return
210 + }
211 + _, _ = io.Copy(j.out, j.buf)
212 + j.buf.Reset()
213 + j.hostState.clearAfterCleanup()
214 }
215
216 func (j *JobV2) AutoDetection() (err error) {
@@ -334,18 +370,18 @@ func (j *JobV2) runOnce() {
370 sinceLastRun := calcSinceLastRun(curTime, j.prevRun)
371 j.prevRun = curTime
372
337 - ok := j.collectAndEmit(sinceLastRun)
373 + prepared, ok := j.collectAndEmit(sinceLastRun)
374 + if ok && !j.panicked.Load() {
375 + if err := j.finishPreparedEmission(prepared); err != nil {
376 + j.Warningf("finalize emission failed: %v", err)
377 + ok = false
378 + }
379 + }
380 if ok {
381 j.retries.Store(0)
382 } else {
383 j.retries.Add(1)
384 }
343 -
344 - // Never flush buffered output from failed or panicked cycles:
345 - // a panic can leave partial protocol lines in the buffer.
346 - if ok && !j.panicked.Load() {
347 - _, _ = io.Copy(j.out, j.buf)
348 - }
385 j.buf.Reset()
386 }
387
@@ -365,13 +401,16 @@ func (j *JobV2) applyPendingVnodeUpdate() {
401 j.vnodeMu.Lock()
402 j.vnode = *next
403 j.vnodeMu.Unlock()
404 + j.hostState.invalidateDefine()
405 default:
406 }
407 }
408
372 -func (j *JobV2) collectAndEmit(sinceLastRun int) bool {
409 +func (j *JobV2) collectAndEmit(sinceLastRun int) (prepared jobV2PreparedEmission, ok bool) {
410 j.panicked.Store(false)
411 cycleOpen := false
412 + var attempt chartengine.PlanAttempt
413 + attemptPending := false
414
415 defer func() {
416 if r := recover(); r != nil {
@@ -382,6 +421,9 @@ func (j *JobV2) collectAndEmit(sinceLastRun int) bool {
421 j.cycle.AbortCycle()
422 }()
423 }
424 + if attemptPending {
425 + attempt.Abort()
426 + }
427 j.panicked.Store(true)
428 j.Errorf("PANIC: %v", r)
429 if logger.Level.Enabled(slog.LevelDebug) {
@@ -396,18 +438,56 @@ func (j *JobV2) collectAndEmit(sinceLastRun int) bool {
438 j.cycle.AbortCycle()
439 cycleOpen = false
440 j.Warningf("collect failed: %v", err)
399 - return false
441 + return jobV2PreparedEmission{}, false
442 }
443 j.cycle.CommitCycleSuccess()
444 cycleOpen = false
445
404 - plan, err := j.engine.BuildPlan(j.store.Read(metrix.ReadRaw(), metrix.ReadFlatten()))
446 + vnode := j.currentVnode()
447 + decision, err := j.hostState.prepareEmission(vnode)
448 + if err != nil {
449 + j.Warningf("prepare host state failed: %v", err)
450 + return jobV2PreparedEmission{}, false
451 + }
452 + if decision.needEngineReload {
453 + j.engine.ResetMaterialized()
454 + j.hostState.onEngineReload(decision.targetHost)
455 + }
456 + attempt, err = j.engine.PreparePlan(j.store.Read(metrix.ReadRaw(), metrix.ReadFlatten()))
457 if err != nil {
458 j.Warningf("build plan failed: %v", err)
407 - return false
459 + return jobV2PreparedEmission{}, false
460 + }
461 + attemptPending = true
462 + plan := attempt.Plan()
463 +
464 + env := j.emitEnv(sinceLastRun, decision)
465 + if err := chartemit.ApplyPlan(j.api, plan, env); err != nil {
466 + attempt.Abort()
467 + attemptPending = false
468 + j.Warningf("apply plan failed: %v", err)
469 + return jobV2PreparedEmission{}, false
470 + }
471 + return jobV2PreparedEmission{
472 + attempt: attempt,
473 + plan: plan,
474 + decision: decision,
475 + }, true
476 +}
477 +
478 +func (j *JobV2) finishPreparedEmission(prepared jobV2PreparedEmission) error {
479 + if j.buf.Len() > 0 {
480 + _, _ = io.Copy(j.out, j.buf)
481 + }
482 + if err := prepared.attempt.Commit(); err != nil {
483 + return err
484 }
485 + j.hostState.commitSuccessfulEmission(prepared.plan, prepared.decision)
486 + return nil
487 +}
488
410 - if err := chartemit.ApplyPlan(j.api, plan, chartemit.EmitEnv{
489 +func (j *JobV2) emitEnv(sinceLastRun int, decision jobV2EmissionDecision) chartemit.EmitEnv {
490 + env := chartemit.EmitEnv{
491 TypeID: j.fullName,
492 UpdateEvery: j.updateEvery,
493 Plugin: j.pluginName,
@@ -415,11 +495,20 @@ func (j *JobV2) collectAndEmit(sinceLastRun int) bool {
495 JobName: j.name,
496 JobLabels: j.labels,
497 MSSinceLast: sinceLastRun,
418 - }); err != nil {
419 - j.Warningf("apply plan failed: %v", err)
420 - return false
498 }
422 - return true
499 + env.HostScope = decision.hostScope
500 + return env
501 +}
502 +
503 +func (j *JobV2) currentVnode() vnodes.VirtualNode {
504 + if j.module != nil {
505 + if vnode := j.module.VirtualNode(); vnode != nil {
506 + return *vnode.Copy()
507 + }
508 + }
509 + j.vnodeMu.RLock()
510 + defer j.vnodeMu.RUnlock()
511 + return *j.vnode.Copy()
512 }
513
514 func (j *JobV2) penalty() int {
src/go/plugin/framework/jobruntime/job_v2_cleanup.go new
+68
@@ -0,0 +1,68 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package jobruntime
4 +
5 +import (
6 + "maps"
7 + "sort"
8 +
9 + "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
10 + "github.com/netdata/netdata/go/plugins/plugin/framework/chartengine"
11 + "github.com/netdata/netdata/go/plugins/plugin/framework/vnodes"
12 +)
13 +
14 +type jobV2CleanupSnapshot struct {
15 + charts map[string]chartengine.ChartMeta
16 + host jobV2HostRef
17 + staleVnodeSuppressed bool
18 +}
19 +
20 +func (s *jobV2HostState) captureCleanupSnapshot(vnode vnodes.VirtualNode) jobV2CleanupSnapshot {
21 + if s == nil {
22 + return jobV2CleanupSnapshot{}
23 + }
24 + host := s.cleanupOwner
25 + return jobV2CleanupSnapshot{
26 + charts: maps.Clone(s.cleanupCharts),
27 + host: host,
28 + staleVnodeSuppressed: shouldSuppressCleanupForStaleVnode(host, vnode),
29 + }
30 +}
31 +
32 +func (s *jobV2HostState) clearAfterCleanup() {
33 + if s == nil {
34 + return
35 + }
36 + clear(s.cleanupCharts)
37 + s.definedHost = jobV2HostRef{}
38 + s.definedInfo = netdataapi.HostInfo{}
39 + s.engineHost = jobV2HostRef{}
40 + s.cleanupOwner = jobV2HostRef{}
41 +}
42 +
43 +func buildJobV2CleanupPlan(charts map[string]chartengine.ChartMeta) chartengine.Plan {
44 + if len(charts) == 0 {
45 + return chartengine.Plan{}
46 + }
47 +
48 + chartIDs := make([]string, 0, len(charts))
49 + for chartID := range charts {
50 + chartIDs = append(chartIDs, chartID)
51 + }
52 + sort.Strings(chartIDs)
53 +
54 + actions := make([]chartengine.EngineAction, 0, len(chartIDs))
55 + for _, chartID := range chartIDs {
56 + actions = append(actions, chartengine.RemoveChartAction{
57 + ChartID: chartID,
58 + Meta: charts[chartID],
59 + })
60 + }
61 + return chartengine.Plan{Actions: actions}
62 +}
63 +
64 +func shouldSuppressCleanupForStaleVnode(cleanupHost jobV2HostRef, vnode vnodes.VirtualNode) bool {
65 + return cleanupHost.isVnode() &&
66 + vnode.GUID == cleanupHost.guid &&
67 + vnode.Labels["_node_stale_after_seconds"] != ""
68 +}
src/go/plugin/framework/jobruntime/job_v2_host_state.go new
+175
@@ -0,0 +1,175 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package jobruntime
4 +
5 +import (
6 + "fmt"
7 + "maps"
8 +
9 + "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
10 + "github.com/netdata/netdata/go/plugins/plugin/framework/chartemit"
11 + "github.com/netdata/netdata/go/plugins/plugin/framework/chartengine"
12 + "github.com/netdata/netdata/go/plugins/plugin/framework/vnodes"
13 +)
14 +
15 +type jobV2HostKind uint8
16 +
17 +const (
18 + jobV2HostUnset jobV2HostKind = iota
19 + jobV2HostGlobal
20 + jobV2HostVnode
21 +)
22 +
23 +type jobV2HostRef struct {
24 + kind jobV2HostKind
25 + guid string
26 +}
27 +
28 +func jobV2HostFromVnode(vnode vnodes.VirtualNode) jobV2HostRef {
29 + if vnode.GUID == "" {
30 + return jobV2HostRef{kind: jobV2HostGlobal}
31 + }
32 + return jobV2HostRef{kind: jobV2HostVnode, guid: vnode.GUID}
33 +}
34 +
35 +func (r jobV2HostRef) isSet() bool { return r.kind != jobV2HostUnset }
36 +func (r jobV2HostRef) isGlobal() bool { return r.kind == jobV2HostGlobal }
37 +func (r jobV2HostRef) isVnode() bool { return r.kind == jobV2HostVnode }
38 +
39 +type jobV2EmissionDecision struct {
40 + targetHost jobV2HostRef
41 + needEngineReload bool
42 + hostScope *chartemit.HostScope
43 + defineEmitted bool
44 + defineInfo netdataapi.HostInfo
45 +}
46 +
47 +type jobV2HostState struct {
48 + definedHost jobV2HostRef
49 + definedInfo netdataapi.HostInfo
50 + engineHost jobV2HostRef
51 + cleanupOwner jobV2HostRef
52 + cleanupCharts map[string]chartengine.ChartMeta
53 +}
54 +
55 +func (s *jobV2HostState) invalidateDefine() {
56 + if s == nil {
57 + return
58 + }
59 + s.definedHost = jobV2HostRef{}
60 + s.definedInfo = netdataapi.HostInfo{}
61 +}
62 +
63 +func (s *jobV2HostState) prepareEmission(vnode vnodes.VirtualNode) (jobV2EmissionDecision, error) {
64 + target := jobV2HostFromVnode(vnode)
65 + decision := jobV2EmissionDecision{
66 + targetHost: target,
67 + needEngineReload: s != nil && s.engineHost.isSet() && s.engineHost != target,
68 + }
69 + if target.isGlobal() {
70 + return decision, nil
71 + }
72 +
73 + info, needDefine, err := s.prepareDefine(vnode, target)
74 + if err != nil {
75 + return jobV2EmissionDecision{}, err
76 + }
77 + scope := &chartemit.HostScope{GUID: target.guid}
78 + if needDefine {
79 + scope.Define = &info
80 + }
81 + decision.hostScope = scope
82 + decision.defineEmitted = needDefine
83 + decision.defineInfo = info
84 + return decision, nil
85 +}
86 +
87 +func (s *jobV2HostState) prepareDefine(vnode vnodes.VirtualNode, target jobV2HostRef) (netdataapi.HostInfo, bool, error) {
88 + info, err := chartemit.PrepareHostInfo(netdataapi.HostInfo{
89 + GUID: vnode.GUID,
90 + Hostname: vnode.Hostname,
91 + Labels: vnode.Labels,
92 + })
93 + if err != nil {
94 + return netdataapi.HostInfo{}, false, err
95 + }
96 + if s != nil && s.definedHost == target && hostInfoEqual(s.definedInfo, info) {
97 + return netdataapi.HostInfo{}, false, nil
98 + }
99 + return info, true, nil
100 +}
101 +
102 +func (s *jobV2HostState) onEngineReload(target jobV2HostRef) {
103 + if s == nil {
104 + return
105 + }
106 + s.engineHost = target
107 +}
108 +
109 +func (s *jobV2HostState) commitSuccessfulEmission(plan chartengine.Plan, decision jobV2EmissionDecision) {
110 + if s == nil || len(plan.Actions) == 0 {
111 + return
112 + }
113 + s.engineHost = decision.targetHost
114 + if decision.defineEmitted {
115 + s.definedHost = decision.targetHost
116 + s.definedInfo = decision.defineInfo
117 + }
118 + if s.cleanupCharts == nil {
119 + s.cleanupCharts = make(map[string]chartengine.ChartMeta)
120 + }
121 +
122 + createCharts := make(map[string]chartengine.ChartMeta)
123 + dimensionOnlyCharts := make(map[string]chartengine.ChartMeta)
124 + removeCharts := make(map[string]struct{})
125 +
126 + for _, action := range plan.Actions {
127 + switch v := action.(type) {
128 + case chartengine.CreateChartAction:
129 + createCharts[v.ChartID] = v.Meta
130 + case chartengine.CreateDimensionAction:
131 + if _, ok := createCharts[v.ChartID]; ok {
132 + continue
133 + }
134 + if _, ok := dimensionOnlyCharts[v.ChartID]; !ok {
135 + dimensionOnlyCharts[v.ChartID] = v.ChartMeta
136 + }
137 + case chartengine.RemoveChartAction:
138 + removeCharts[v.ChartID] = struct{}{}
139 + }
140 + }
141 +
142 + for chartID, meta := range createCharts {
143 + s.cleanupCharts[chartID] = meta
144 + }
145 + for chartID, meta := range dimensionOnlyCharts {
146 + if _, ok := s.cleanupCharts[chartID]; ok {
147 + continue
148 + }
149 + s.cleanupCharts[chartID] = meta
150 + }
151 + for chartID := range removeCharts {
152 + delete(s.cleanupCharts, chartID)
153 + }
154 +
155 + s.cleanupOwner = decision.targetHost
156 +}
157 +
158 +func hostInfoEqual(left, right netdataapi.HostInfo) bool {
159 + return left.GUID == right.GUID &&
160 + left.Hostname == right.Hostname &&
161 + maps.Equal(left.Labels, right.Labels)
162 +}
163 +
164 +func (r jobV2HostRef) String() string {
165 + switch r.kind {
166 + case jobV2HostUnset:
167 + return "unset"
168 + case jobV2HostGlobal:
169 + return "global"
170 + case jobV2HostVnode:
171 + return fmt.Sprintf("vnode(%s)", r.guid)
172 + default:
173 + return "unknown"
174 + }
175 +}
src/go/plugin/framework/jobruntime/job_v2_test.go
+490 -18
@@ -9,6 +9,9 @@ import (
9 "testing"
10 "time"
11
12 + "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
13 + "github.com/netdata/netdata/go/plugins/plugin/framework/chartemit"
14 + "github.com/netdata/netdata/go/plugins/plugin/framework/chartengine"
15 "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
16 "github.com/netdata/netdata/go/plugins/plugin/framework/runtimecomp"
17 "github.com/netdata/netdata/go/plugins/plugin/framework/vnodes"
@@ -26,10 +29,11 @@ type mockModuleV2 struct {
29 collectFunc func(context.Context) error
30 cleanupFunc func(context.Context)
31
29 - store metrix.CollectorStore
30 - template string
31 - cleaned bool
32 - vnode *vnodes.VirtualNode
32 + store metrix.CollectorStore
33 + template string
34 + templateCalls int
35 + cleaned bool
36 + vnode *vnodes.VirtualNode
37 }
38
39 type mockRuntimeComponentService struct {
@@ -85,7 +89,10 @@ func (m *mockModuleV2) Cleanup(ctx context.Context) {
89 func (m *mockModuleV2) Configuration() any { return nil }
90 func (m *mockModuleV2) VirtualNode() *vnodes.VirtualNode { return m.vnode }
91 func (m *mockModuleV2) MetricStore() metrix.CollectorStore { return m.store }
88 -func (m *mockModuleV2) ChartTemplateYAML() string { return m.template }
92 +func (m *mockModuleV2) ChartTemplateYAML() string {
93 + m.templateCalls++
94 + return m.template
95 +}
96
97 func newTestJobV2(mod collectorapi.CollectorV2, out *bytes.Buffer) *JobV2 {
98 return NewJobV2(JobV2Config{
@@ -174,7 +181,10 @@ func TestJobV2Scenarios(t *testing.T) {
181 require.NotNil(t, job.store)
182 require.NotNil(t, job.cycle)
183 require.NotNil(t, job.engine)
177 - _, err := job.engine.BuildPlan(job.store.Read(metrix.ReadFlatten()))
184 + attempt, err := job.engine.PreparePlan(job.store.Read(metrix.ReadFlatten()))
185 + require.NoError(t, err)
186 + defer attempt.Abort()
187 + err = attempt.Commit()
188 require.NoError(t, err)
189 },
190 },
@@ -206,10 +216,16 @@ func TestJobV2Scenarios(t *testing.T) {
216 job.runOnce()
217
218 wire := out.String()
209 - assert.Contains(t, wire, "CHART 'module_job.workers_busy'")
210 - assert.Contains(t, wire, "CLABEL '_collect_job' 'job' '1'")
211 - assert.Contains(t, wire, "BEGIN 'module_job.workers_busy'")
212 - assert.Contains(t, wire, "SET 'busy' = 7")
219 + assert.Contains(t, wire, `HOST ''
220 +
221 +CHART 'module_job.workers_busy' '' 'Workers Busy' 'workers' 'Workers' 'workers_busy' 'line' '0' '1' '' 'plugin' 'module'
222 +CLABEL 'instance' 'localhost' '2'
223 +CLABEL '_collect_job' 'job' '1'
224 +CLABEL_COMMIT
225 +DIMENSION 'busy' 'busy' 'absolute' '1' '1' ''
226 +BEGIN 'module_job.workers_busy'
227 +SET 'busy' = 7
228 +END`)
229 assert.False(t, job.Panicked())
230 },
231 },
@@ -293,14 +309,29 @@ func TestJobV2Scenarios(t *testing.T) {
309 job.runOnce()
310
311 wire := out.String()
296 - assert.Contains(t, wire, "CHART 'module_job.win_nic_traffic_eth0'")
297 - assert.Contains(t, wire, "CHART 'module_job.win_nic_traffic_eth1'")
298 - assert.Contains(t, wire, "BEGIN 'module_job.win_nic_traffic_eth0'")
299 - assert.Contains(t, wire, "SET 'received' = 100")
300 - assert.Contains(t, wire, "SET 'sent' = 80")
301 - assert.Contains(t, wire, "BEGIN 'module_job.win_nic_traffic_eth1'")
302 - assert.Contains(t, wire, "SET 'received' = 50")
303 - assert.Contains(t, wire, "SET 'sent' = 40")
312 + assert.Contains(t, wire, `CHART 'module_job.win_nic_traffic_eth0' '' 'NIC traffic' 'bytes/s' 'Net' 'nic_traffic' 'line' '0' '1' '' 'plugin' 'module'
313 +CLABEL 'instance' 'localhost' '2'
314 +CLABEL 'nic' 'eth0' '1'
315 +CLABEL '_collect_job' 'job' '1'
316 +CLABEL_COMMIT
317 +DIMENSION 'received' 'received' 'incremental' '1' '1' ''
318 +DIMENSION 'sent' 'sent' 'incremental' '1' '1' ''
319 +CHART 'module_job.win_nic_traffic_eth1' '' 'NIC traffic' 'bytes/s' 'Net' 'nic_traffic' 'line' '0' '1' '' 'plugin' 'module'
320 +CLABEL 'instance' 'localhost' '2'
321 +CLABEL 'nic' 'eth1' '1'
322 +CLABEL '_collect_job' 'job' '1'
323 +CLABEL_COMMIT
324 +DIMENSION 'received' 'received' 'incremental' '1' '1' ''
325 +DIMENSION 'sent' 'sent' 'incremental' '1' '1' ''
326 +BEGIN 'module_job.win_nic_traffic_eth0'
327 +SET 'received' = 100
328 +SET 'sent' = 80
329 +END
330 +
331 +BEGIN 'module_job.win_nic_traffic_eth1'
332 +SET 'received' = 50
333 +SET 'sent' = 40
334 +END`)
335 },
336 },
337 "runtime component registers on successful autodetection": {
@@ -698,6 +729,447 @@ func TestJobV2_StartMarksNotRunningBeforeCleanup(t *testing.T) {
729 }
730 }
731
732 +func TestJobV2VnodeEmissionLifecycle(t *testing.T) {
733 + store := metrix.NewCollectorStore()
734 + current := 1.0
735 + mod := &mockModuleV2{
736 + store: store,
737 + template: chartTemplateV2(),
738 + collectFunc: func(context.Context) error {
739 + store.Write().SnapshotMeter("apache").Gauge("workers_busy").Observe(current)
740 + return nil
741 + },
742 + }
743 +
744 + var out bytes.Buffer
745 + job := newTestJobV2WithVnode(mod, &out, vnodes.VirtualNode{
746 + Hostname: "node-host",
747 + GUID: "node-guid",
748 + Labels: map[string]string{
749 + "region": "eu'\n",
750 + },
751 + })
752 + require.NoError(t, job.AutoDetection())
753 +
754 + job.runOnce()
755 + wire := out.String()
756 + assert.Contains(t, wire, `HOST_DEFINE 'node-guid' 'node-host'
757 +HOST_LABEL '_hostname' 'node-host'
758 +HOST_LABEL 'region' 'eu '
759 +HOST_DEFINE_END
760 +
761 +HOST 'node-guid'
762 +
763 +CHART 'module_job.workers_busy'`)
764 + assert.NotContains(t, wire, "HOST ''")
765 +
766 + out.Reset()
767 + current = 2
768 + job.runOnce()
769 + wire = out.String()
770 + assert.Contains(t, wire, `HOST 'node-guid'
771 +
772 +BEGIN 'module_job.workers_busy'`)
773 + assert.NotContains(t, wire, "HOST_DEFINE 'node-guid' 'node-host'")
774 +
775 + out.Reset()
776 + job.UpdateVnode(&vnodes.VirtualNode{
777 + Hostname: "node-host-2",
778 + GUID: "node-guid-2",
779 + })
780 + current = 3
781 + job.runOnce()
782 + wire = out.String()
783 + assert.Contains(t, wire, `HOST_DEFINE 'node-guid-2' 'node-host-2'
784 +HOST_LABEL '_hostname' 'node-host-2'
785 +HOST_DEFINE_END
786 +
787 +HOST 'node-guid-2'
788 +
789 +CHART 'module_job.workers_busy'`)
790 +}
791 +
792 +func TestJobV2ModuleOwnedVnodeSameGUIDMetadataRefresh(t *testing.T) {
793 + cases := map[string]struct {
794 + mutate func(*vnodes.VirtualNode)
795 + wantDefine bool
796 + wantDefineWire string
797 + wantInfo netdataapi.HostInfo
798 + }{
799 + "unchanged metadata does not redefine": {
800 + mutate: func(*vnodes.VirtualNode) {},
801 + wantDefine: false,
802 + wantInfo: netdataapi.HostInfo{
803 + GUID: "node-guid",
804 + Hostname: "node-host-a",
805 + Labels: map[string]string{
806 + "_hostname": "node-host-a",
807 + "region": "eu",
808 + },
809 + },
810 + },
811 + "hostname change redefines same guid": {
812 + mutate: func(vnode *vnodes.VirtualNode) {
813 + vnode.Hostname = "node-host-b"
814 + },
815 + wantDefine: true,
816 + wantDefineWire: `HOST_DEFINE 'node-guid' 'node-host-b'
817 +HOST_LABEL '_hostname' 'node-host-b'
818 +HOST_LABEL 'region' 'eu'
819 +HOST_DEFINE_END
820 +
821 +HOST 'node-guid'
822 +
823 +BEGIN 'module_job.workers_busy'`,
824 + wantInfo: netdataapi.HostInfo{
825 + GUID: "node-guid",
826 + Hostname: "node-host-b",
827 + Labels: map[string]string{
828 + "_hostname": "node-host-b",
829 + "region": "eu",
830 + },
831 + },
832 + },
833 + "label change redefines same guid": {
834 + mutate: func(vnode *vnodes.VirtualNode) {
835 + vnode.Labels["region"] = "us"
836 + },
837 + wantDefine: true,
838 + wantDefineWire: `HOST_DEFINE 'node-guid' 'node-host-a'
839 +HOST_LABEL '_hostname' 'node-host-a'
840 +HOST_LABEL 'region' 'us'
841 +HOST_DEFINE_END
842 +
843 +HOST 'node-guid'
844 +
845 +BEGIN 'module_job.workers_busy'`,
846 + wantInfo: netdataapi.HostInfo{
847 + GUID: "node-guid",
848 + Hostname: "node-host-a",
849 + Labels: map[string]string{
850 + "_hostname": "node-host-a",
851 + "region": "us",
852 + },
853 + },
854 + },
855 + }
856 +
857 + for name, tc := range cases {
858 + t.Run(name, func(t *testing.T) {
859 + store := metrix.NewCollectorStore()
860 + current := 1.0
861 + modVnode := &vnodes.VirtualNode{
862 + Hostname: "node-host-a",
863 + GUID: "node-guid",
864 + Labels: map[string]string{
865 + "region": "eu",
866 + },
867 + }
868 + mod := &mockModuleV2{
869 + store: store,
870 + template: chartTemplateV2(),
871 + vnode: modVnode,
872 + collectFunc: func(context.Context) error {
873 + store.Write().SnapshotMeter("apache").Gauge("workers_busy").Observe(current)
874 + return nil
875 + },
876 + }
877 +
878 + var out bytes.Buffer
879 + job := newTestJobV2WithVnode(mod, &out, *modVnode.Copy())
880 + require.NoError(t, job.AutoDetection())
881 +
882 + initialInfo, err := chartemit.PrepareHostInfo(netdataapi.HostInfo{
883 + GUID: "node-guid",
884 + Hostname: "node-host-a",
885 + Labels: map[string]string{
886 + "region": "eu",
887 + },
888 + })
889 + require.NoError(t, err)
890 +
891 + job.runOnce()
892 + require.Equal(t, initialInfo, job.hostState.definedInfo)
893 +
894 + out.Reset()
895 + tc.mutate(modVnode)
896 + current = 2
897 +
898 + job.runOnce()
899 + wire := out.String()
900 + if tc.wantDefine {
901 + assert.Contains(t, wire, tc.wantDefineWire)
902 + } else {
903 + assert.NotContains(t, wire, `HOST_DEFINE 'node-guid'`)
904 + assert.Contains(t, wire, `HOST 'node-guid'
905 +
906 +BEGIN 'module_job.workers_busy'`)
907 + }
908 +
909 + expectedInfo, err := chartemit.PrepareHostInfo(tc.wantInfo)
910 + require.NoError(t, err)
911 + assert.Equal(t, expectedInfo, job.hostState.definedInfo)
912 + })
913 + }
914 +}
915 +
916 +func TestJobV2EmptyPlanDoesNotMarkVnodeDefined(t *testing.T) {
917 + store := metrix.NewCollectorStore()
918 + emitValue := false
919 + mod := &mockModuleV2{
920 + store: store,
921 + template: chartTemplateV2(),
922 + collectFunc: func(context.Context) error {
923 + if emitValue {
924 + store.Write().SnapshotMeter("apache").Gauge("workers_busy").Observe(1)
925 + }
926 + return nil
927 + },
928 + }
929 +
930 + var out bytes.Buffer
931 + job := newTestJobV2WithVnode(mod, &out, vnodes.VirtualNode{
932 + Hostname: "node-host",
933 + GUID: "node-guid",
934 + })
935 + require.NoError(t, job.AutoDetection())
936 +
937 + job.runOnce()
938 + assert.Equal(t, "", out.String())
939 + assert.False(t, job.hostState.definedHost.isSet())
940 +
941 + emitValue = true
942 + job.runOnce()
943 + assert.Contains(t, out.String(), `HOST_DEFINE 'node-guid' 'node-host'`)
944 + assert.Equal(t, jobV2HostRef{kind: jobV2HostVnode, guid: "node-guid"}, job.hostState.definedHost)
945 +}
946 +
947 +func TestJobV2CleanupUsesLastSuccessfulHostAfterFailedHostSwitch(t *testing.T) {
948 + store := metrix.NewCollectorStore()
949 + current := 1.0
950 + failCollect := false
951 + mod := &mockModuleV2{
952 + store: store,
953 + template: chartTemplateV2(),
954 + collectFunc: func(context.Context) error {
955 + if failCollect {
956 + return errors.New("collect failed")
957 + }
958 + store.Write().SnapshotMeter("apache").Gauge("workers_busy").Observe(current)
959 + return nil
960 + },
961 + }
962 +
963 + var out bytes.Buffer
964 + job := newTestJobV2WithVnode(mod, &out, vnodes.VirtualNode{
965 + Hostname: "node-host-a",
966 + GUID: "node-guid-a",
967 + })
968 + require.NoError(t, job.AutoDetection())
969 +
970 + job.runOnce()
971 + out.Reset()
972 +
973 + failCollect = true
974 + job.UpdateVnode(&vnodes.VirtualNode{
975 + Hostname: "node-host-b",
976 + GUID: "node-guid-b",
977 + })
978 + job.runOnce()
979 + assert.Equal(t, "", out.String())
980 +
981 + job.Cleanup()
982 +
983 + wire := out.String()
984 + assert.Contains(t, wire, `HOST 'node-guid-a'
985 +
986 +CHART 'module_job.workers_busy' '' 'Workers Busy' 'workers' 'Workers' 'workers_busy' 'line' '0' '1' 'obsolete' 'plugin' 'module'`)
987 + assert.NotContains(t, wire, "HOST 'node-guid-b'")
988 + assert.Empty(t, job.hostState.cleanupCharts)
989 + assert.False(t, job.hostState.cleanupOwner.isSet())
990 +}
991 +
992 +func TestJobV2EmptyHostSwitchDoesNotKeepReloadingEngine(t *testing.T) {
993 + store := metrix.NewCollectorStore()
994 + emitValue := true
995 + mod := &mockModuleV2{
996 + store: store,
997 + template: chartTemplateV2(),
998 + collectFunc: func(context.Context) error {
999 + if emitValue {
1000 + store.Write().SnapshotMeter("apache").Gauge("workers_busy").Observe(1)
1001 + }
1002 + return nil
1003 + },
1004 + }
1005 +
1006 + var out bytes.Buffer
1007 + job := newTestJobV2WithVnode(mod, &out, vnodes.VirtualNode{
1008 + Hostname: "node-host-a",
1009 + GUID: "node-guid-a",
1010 + })
1011 + require.NoError(t, job.AutoDetection())
1012 + require.Equal(t, 1, mod.templateCalls)
1013 +
1014 + job.runOnce()
1015 + require.Equal(t, 1, mod.templateCalls)
1016 + require.Equal(t, jobV2HostRef{kind: jobV2HostVnode, guid: "node-guid-a"}, job.hostState.engineHost)
1017 + require.Equal(t, jobV2HostRef{kind: jobV2HostVnode, guid: "node-guid-a"}, job.hostState.cleanupOwner)
1018 +
1019 + out.Reset()
1020 + emitValue = false
1021 + job.UpdateVnode(&vnodes.VirtualNode{
1022 + Hostname: "node-host-b",
1023 + GUID: "node-guid-b",
1024 + })
1025 + job.runOnce()
1026 + assert.Equal(t, "", out.String())
1027 + require.Equal(t, 1, mod.templateCalls)
1028 + require.Equal(t, jobV2HostRef{kind: jobV2HostVnode, guid: "node-guid-b"}, job.hostState.engineHost)
1029 + require.Equal(t, jobV2HostRef{kind: jobV2HostVnode, guid: "node-guid-a"}, job.hostState.cleanupOwner)
1030 +
1031 + out.Reset()
1032 + job.runOnce()
1033 + assert.Equal(t, "", out.String())
1034 + assert.Equal(t, 1, mod.templateCalls)
1035 + assert.Equal(t, jobV2HostRef{kind: jobV2HostVnode, guid: "node-guid-b"}, job.hostState.engineHost)
1036 + assert.Equal(t, jobV2HostRef{kind: jobV2HostVnode, guid: "node-guid-a"}, job.hostState.cleanupOwner)
1037 +}
1038 +
1039 +func TestJobV2CleanupDoesNotSuppressGlobalCleanupForDifferentStaleVnode(t *testing.T) {
1040 + store := metrix.NewCollectorStore()
1041 + failCollect := false
1042 + mod := &mockModuleV2{
1043 + store: store,
1044 + template: chartTemplateV2(),
1045 + collectFunc: func(context.Context) error {
1046 + if failCollect {
1047 + return errors.New("collect failed")
1048 + }
1049 + store.Write().SnapshotMeter("apache").Gauge("workers_busy").Observe(1)
1050 + return nil
1051 + },
1052 + }
1053 +
1054 + var out bytes.Buffer
1055 + job := newTestJobV2(mod, &out)
1056 + require.NoError(t, job.AutoDetection())
1057 +
1058 + job.runOnce()
1059 + out.Reset()
1060 +
1061 + failCollect = true
1062 + job.UpdateVnode(&vnodes.VirtualNode{
1063 + Hostname: "node-host-b",
1064 + GUID: "node-guid-b",
1065 + Labels: map[string]string{
1066 + "_node_stale_after_seconds": "60",
1067 + },
1068 + })
1069 + job.runOnce()
1070 + assert.Equal(t, "", out.String())
1071 +
1072 + job.Cleanup()
1073 +
1074 + wire := out.String()
1075 + assert.Contains(t, wire, `HOST ''
1076 +
1077 +CHART 'module_job.workers_busy' '' 'Workers Busy' 'workers' 'Workers' 'workers_busy' 'line' '0' '1' 'obsolete' 'plugin' 'module'`)
1078 + assert.NotContains(t, wire, "HOST 'node-guid-b'")
1079 +}
1080 +
1081 +func TestJobV2CleanupUsesPreModuleCleanupSnapshotForStaleSuppression(t *testing.T) {
1082 + store := metrix.NewCollectorStore()
1083 + modVnode := &vnodes.VirtualNode{
1084 + Hostname: "node-host-a",
1085 + GUID: "node-guid-a",
1086 + }
1087 + mod := &mockModuleV2{
1088 + store: store,
1089 + template: chartTemplateV2(),
1090 + vnode: modVnode,
1091 + collectFunc: func(context.Context) error {
1092 + store.Write().SnapshotMeter("apache").Gauge("workers_busy").Observe(1)
1093 + return nil
1094 + },
1095 + }
1096 + mod.cleanupFunc = func(context.Context) {
1097 + *modVnode = vnodes.VirtualNode{
1098 + Hostname: "node-host-b",
1099 + GUID: "node-guid-b",
1100 + }
1101 + }
1102 +
1103 + var out bytes.Buffer
1104 + job := newTestJobV2WithVnode(mod, &out, *modVnode.Copy())
1105 + require.NoError(t, job.AutoDetection())
1106 +
1107 + job.runOnce()
1108 + out.Reset()
1109 +
1110 + modVnode.Labels = map[string]string{
1111 + "_node_stale_after_seconds": "60",
1112 + }
1113 +
1114 + job.Cleanup()
1115 +
1116 + assert.Equal(t, "", out.String())
1117 + assert.True(t, mod.cleaned)
1118 +}
1119 +
1120 +func TestJobV2CleanupNoSuccessfulEmissionsIsNoOp(t *testing.T) {
1121 + mod := &mockModuleV2{
1122 + store: metrix.NewCollectorStore(),
1123 + template: chartTemplateV2(),
1124 + }
1125 +
1126 + var out bytes.Buffer
1127 + job := newTestJobV2(mod, &out)
1128 + require.NoError(t, job.AutoDetection())
1129 +
1130 + job.Cleanup()
1131 +
1132 + assert.Equal(t, "", out.String())
1133 + assert.True(t, mod.cleaned)
1134 +}
1135 +
1136 +func TestJobV2CleanupTrackerUsesEffectiveEmittedChartSet(t *testing.T) {
1137 + meta := chartengine.ChartMeta{
1138 + Title: "Workers Busy",
1139 + Family: "Workers",
1140 + Context: "workers_busy",
1141 + Units: "workers",
1142 + Type: chartengine.ChartTypeLine,
1143 + }
1144 +
1145 + job := &JobV2{}
1146 + decision := jobV2EmissionDecision{targetHost: jobV2HostRef{kind: jobV2HostGlobal}}
1147 + job.hostState.commitSuccessfulEmission(chartengine.Plan{
1148 + Actions: []chartengine.EngineAction{
1149 + chartengine.CreateDimensionAction{
1150 + ChartID: "workers_busy",
1151 + ChartMeta: meta,
1152 + Name: "busy",
1153 + },
1154 + },
1155 + }, decision)
1156 +
1157 + require.Len(t, job.hostState.cleanupCharts, 1)
1158 + assert.Equal(t, meta, job.hostState.cleanupCharts["workers_busy"])
1159 + assert.Equal(t, jobV2HostRef{kind: jobV2HostGlobal}, job.hostState.cleanupOwner)
1160 +
1161 + job.hostState.commitSuccessfulEmission(chartengine.Plan{
1162 + Actions: []chartengine.EngineAction{
1163 + chartengine.RemoveChartAction{
1164 + ChartID: "workers_busy",
1165 + Meta: meta,
1166 + },
1167 + },
1168 + }, decision)
1169 +
1170 + assert.Empty(t, job.hostState.cleanupCharts)
1171 +}
1172 +
1173 func TestJobV2StopBeforeStartDoesNotBlock(t *testing.T) {
1174 job := NewJobV2(JobV2Config{
1175 PluginName: pluginName,
src/go/plugin/go.d/pkg/collecttest/collecttest.go
+11 -1
@@ -134,7 +134,17 @@ func buildPlanFromTemplate(templateYAML string, revision uint64, reader metrix.R
134 if err := engine.LoadYAML([]byte(templateYAML), revision); err != nil {
135 return chartengine.Plan{}, err
136 }
137 - return engine.BuildPlan(reader)
137 + attempt, err := engine.PreparePlan(reader)
138 + if err != nil {
139 + return chartengine.Plan{}, err
140 + }
141 + defer attempt.Abort()
142 +
143 + plan := attempt.Plan()
144 + if err := attempt.Commit(); err != nil {
145 + return chartengine.Plan{}, err
146 + }
147 + return plan, nil
148 }
149
150 type planFilter struct {