@cryptotaxi247 / netdata-1 / commits / 67f60fd4c

fix(go/plugin/framework/chartengine): decouple runtime build-cycle dedupe from LastSuccessSeq (#21851)

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Ilya Mashchenko committed Mar 1, 2026 at 17:27 UTC 67f60fd4c1117330487a04e9e5ddf3b4feecaa88
32 files changed +859 -385
src/go/pkg/metrix/seeded.go new
+20
@@ -0,0 +1,20 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package metrix
4 +
5 +// SeededGauge declares a stateful gauge and seeds it with zero immediately.
6 +func SeededGauge(m StatefulMeter, name string, opts ...InstrumentOption) StatefulGauge {
7 + g := m.Gauge(name, opts...)
8 + g.Set(0)
9 + return g
10 +}
11 +
12 +// SeededCounter declares a stateful counter and seeds it with zero immediately.
13 +//
14 +// Note: Add(0) advances per-series counter sequence bookkeeping, which is
15 +// harmless for value semantics and ensures a visible committed zero series.
16 +func SeededCounter(m StatefulMeter, name string, opts ...InstrumentOption) StatefulCounter {
17 + c := m.Counter(name, opts...)
18 + c.Add(0)
19 + return c
20 +}
src/go/pkg/metrix/seeded_test.go new
+85
@@ -0,0 +1,85 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package metrix
4 +
5 +import (
6 + "testing"
7 + "time"
8 +)
9 +
10 +func TestSeededHelperScenarios(t *testing.T) {
11 + tests := map[string]struct {
12 + run func(t *testing.T)
13 + }{
14 + "SeededGauge creates visible zero-valued series": {
15 + run: func(t *testing.T) {
16 + s := NewRuntimeStore()
17 + m := s.Write().StatefulMeter("runtime")
18 + _ = SeededGauge(m, "queue_depth")
19 +
20 + mustValue(t, s.Read(ReadRaw()), "runtime.queue_depth", nil, 0)
21 + },
22 + },
23 + "SeededCounter creates visible zero-valued series without initial delta": {
24 + run: func(t *testing.T) {
25 + s := NewRuntimeStore()
26 + m := s.Write().StatefulMeter("runtime")
27 + _ = SeededCounter(m, "jobs_total")
28 +
29 + mustValue(t, s.Read(ReadRaw()), "runtime.jobs_total", nil, 0)
30 + mustNoDelta(t, s.Read(ReadRaw()), "runtime.jobs_total", nil)
31 + },
32 + },
33 + "SeededCounter accumulates normally after seed": {
34 + run: func(t *testing.T) {
35 + s := NewRuntimeStore()
36 + m := s.Write().StatefulMeter("runtime")
37 + c := SeededCounter(m, "events_total")
38 +
39 + c.Add(5)
40 + mustValue(t, s.Read(ReadRaw()), "runtime.events_total", nil, 5)
41 + mustDelta(t, s.Read(ReadRaw()), "runtime.events_total", nil, 5)
42 + },
43 + },
44 + "Seeded helpers preserve meter labels": {
45 + run: func(t *testing.T) {
46 + s := NewRuntimeStore()
47 + m := s.Write().StatefulMeter("runtime").WithLabels(Label{Key: "component", Value: "functions"})
48 + _ = SeededGauge(m, "invocations_active")
49 + _ = SeededCounter(m, "calls_total")
50 +
51 + labels := Labels{"component": "functions"}
52 + mustValue(t, s.Read(ReadRaw()), "runtime.invocations_active", labels, 0)
53 + mustValue(t, s.Read(ReadRaw()), "runtime.calls_total", labels, 0)
54 + },
55 + },
56 + "Seeded series can be evicted by TTL when compaction is triggered later": {
57 + run: func(t *testing.T) {
58 + s := NewRuntimeStore()
59 + view := runtimeStoreViewForTest(t, s)
60 + now := time.Unix(1_700_000_000, 0)
61 + view.backend.now = func() time.Time { return now }
62 + view.backend.retention = runtimeRetentionPolicy{
63 + ttl: 5 * time.Second,
64 + maxSeries: 0,
65 + }
66 + view.backend.compaction = runtimeCompactionPolicy{
67 + maxOverlayDepth: 1,
68 + maxOverlayWrites: 1,
69 + }
70 +
71 + m := s.Write().StatefulMeter("runtime")
72 + _ = SeededCounter(m, "stale_total")
73 + mustValue(t, s.Read(ReadRaw()), "runtime.stale_total", nil, 0)
74 +
75 + now = now.Add(6 * time.Second)
76 + SeededGauge(m, "trigger")
77 + mustNoValue(t, s.Read(ReadRaw()), "runtime.stale_total", nil)
78 + },
79 + },
80 + }
81 +
82 + for name, tc := range tests {
83 + t.Run(name, tc.run)
84 + }
85 +}
src/go/plugin/agent/runtimechartemit/components.go
+1 -4
@@ -145,16 +145,13 @@ func normalizeComponent(cfg ComponentConfig, pluginName string) (componentSpec,
145 JobName: firstNotEmpty(strings.TrimSpace(cfg.JobName), name),
146 JobLabels: cloneStringMap(cfg.JobLabels),
147 }
148 - autogen := cfg.Autogen
149 - // Chartengine autogen type.id budget must use the actual emitted type.id.
150 - autogen.TypeID = typeID
148
149 return componentSpec{
150 Name: name,
151 Store: cfg.Store,
152 TemplateYAML: append([]byte(nil), templateYAML...),
153 UpdateEvery: updateEvery,
157 - Autogen: autogen,
154 + Autogen: cfg.Autogen,
155 EmitEnv: env,
156 }, nil
157 }
src/go/plugin/agent/runtimechartemit/job.go
+3 -1
@@ -188,7 +188,9 @@ func (j *runtimeMetricsJob) ensureComponent(spec componentSpec) (*runtimeCompone
188 engine, err := chartengine.New(
189 chartengine.WithRuntimeStore(nil), // Two-engine policy: observer engine has no self-metrics.
190 chartengine.WithSeriesSelectionAllVisible(),
191 - chartengine.WithAutogenPolicy(spec.Autogen),
191 + chartengine.WithRuntimePlannerMode(),
192 + chartengine.WithEmitTypeIDBudgetPrefix(spec.EmitEnv.TypeID),
193 + chartengine.WithEnginePolicy(chartengine.EnginePolicy{Autogen: &spec.Autogen}),
194 chartengine.WithLogger(engineLog),
195 )
196 if err != nil {
src/go/plugin/agent/runtimechartemit/service_test.go
+30
@@ -152,6 +152,36 @@ func TestRuntimeMetricsJobScenarios(t *testing.T) {
152 assert.Contains(t, result, "DIMENSION 'b' 'b' 'absolute'")
153 },
154 },
155 + "runtime job keeps emitting on no-write ticks": {
156 + run: func(t *testing.T) {
157 + reg := newComponentRegistry()
158 + store := metrix.NewRuntimeStore()
159 + store.Write().StatefulMeter("component").Gauge("load").Set(7)
160 +
161 + reg.upsert(componentSpec{
162 + Name: "component",
163 + Store: store,
164 + TemplateYAML: []byte(runtimeGaugeTemplateYAML()),
165 + UpdateEvery: 1,
166 + EmitEnv: chartemit.EmitEnv{
167 + TypeID: "netdata.go.d.internal.component",
168 + UpdateEvery: 1,
169 + Plugin: "go.d",
170 + Module: "internal",
171 + JobName: "component",
172 + },
173 + })
174 +
175 + var out bytes.Buffer
176 + job := newRuntimeMetricsJob(&out, reg, nil)
177 + job.runOnce(1)
178 + require.Contains(t, out.String(), "BEGIN")
179 +
180 + out.Reset()
181 + job.runOnce(2)
182 + assert.Contains(t, out.String(), "BEGIN")
183 + },
184 + },
185 "runtime job emits obsolete chart when component is removed": {
186 run: func(t *testing.T) {
187 reg := newComponentRegistry()
src/go/plugin/framework/chartengine/README.md
+4 -2
@@ -38,6 +38,8 @@ For `ModuleV2` collectors, the runtime integration expects:
38 | `WithEnginePolicy(...)` | Configure selector + autogen behavior |
39 | `WithRuntimeStore(...)` | Override/disable self-metrics store |
40 | `WithSeriesSelectionAllVisible()` | Process all visible series instead of filtering to latest successful collect cycle. Intended for runtime/internal stores that commit immediately (no cycle boundaries). |
41 +| `WithEmitTypeIDBudgetPrefix(...)` | Set the effective type-id prefix used by autogen budget checks |
42 +| `WithRuntimePlannerMode(...)` | Enable runtime planner mode with no-write-tick semantics, for jobs/tests that drive planning directly from runtime metrics instead of collect-cycle boundaries. |
43
44 ## End-to-End Example (Single Flow)
45
@@ -50,7 +52,7 @@ meter.Counter("requests_total").ObserveTotal(100)
52 // 2) Engine loads chart template.
53 engine, err := chartengine.New(
54 chartengine.WithEnginePolicy(chartengine.EnginePolicy{
53 - Autogen: chartengine.AutogenPolicy{Enabled: false},
55 + Autogen: &chartengine.AutogenPolicy{Enabled: false},
56 }),
57 )
58 // handle err
@@ -160,7 +162,7 @@ Default lifecycle policy when template omits lifecycle:
162 |-----------------------|------------------------------------------------------------------------------------------------------------------------------------------------|
163 | Trigger | Unmatched series only when autogen is enabled |
164 | Metric metadata usage | Uses `metrix.MetricMeta` hints for title/family/unit where allowed |
163 -| Type ID budget | Enforced via `AutogenPolicy.MaxTypeIDLen` (`type.id` length guard) |
165 +| Type ID budget | Enforced via `AutogenPolicy.MaxTypeIDLen` + effective emit type-id prefix (`WithEmitTypeIDBudgetPrefix(...)`) |
166 | Lifecycle | Autogen applies `ExpireAfterSuccessCycles` to **both** chart and dimension expiry (unlike template lifecycle where they default independently) |
167
168 ## Runtime Metrics
src/go/plugin/framework/chartengine/autogen.go
+30 -17
@@ -35,12 +35,14 @@ type autogenSourceBuilder func(
35 labels metrix.LabelView,
36 meta metrix.SeriesMeta,
37 policy AutogenPolicy,
38 + typeIDPrefix string,
39 ) (autogenRoute, bool, error)
40
41 type autogenRoleBuilder func(
42 metricName string,
43 labels metrix.LabelView,
44 policy AutogenPolicy,
45 + typeIDPrefix string,
46 ) (autogenRoute, bool, error)
47
48 var autogenSourceBuilders = map[metrix.MetricKind]autogenSourceBuilder{
@@ -75,7 +77,7 @@ func (e *Engine) resolveAutogenRoute(
77 return nil, false, nil
78 }
79
78 - route, ok, err := buildAutogenRoute(metricName, labels, meta, policy)
80 + route, ok, err := buildAutogenRoute(metricName, labels, meta, policy, e.state.cfg.autogenTypeID)
81 if err != nil {
82 return nil, false, err
83 }
@@ -123,6 +125,7 @@ func buildAutogenRoute(
125 labels metrix.LabelView,
126 meta metrix.SeriesMeta,
127 policy AutogenPolicy,
128 + typeIDPrefix string,
129 ) (autogenRoute, bool, error) {
130 if strings.TrimSpace(metricName) == "" {
131 return autogenRoute{}, false, nil
@@ -131,7 +134,7 @@ func buildAutogenRoute(
134 if knownBuilder, ok := autogenSourceBuilders[meta.SourceKind]; ok {
135 builder = knownBuilder
136 }
134 - return builder(metricName, labels, meta, policy)
137 + return builder(metricName, labels, meta, policy, typeIDPrefix)
138 }
139
140 func autogenMetricMeta(reader metrix.Reader, metricName string, meta metrix.SeriesMeta) (metrix.MetricMeta, bool) {
@@ -231,12 +234,13 @@ func buildHistogramAutogenRoute(
234 labels metrix.LabelView,
235 meta metrix.SeriesMeta,
236 policy AutogenPolicy,
237 + typeIDPrefix string,
238 ) (autogenRoute, bool, error) {
239 builder, ok := histogramRoleBuilders[meta.FlattenRole]
240 if !ok {
241 return autogenRoute{}, false, nil
242 }
239 - return builder(metricName, labels, policy)
243 + return builder(metricName, labels, policy, typeIDPrefix)
244 }
245
246 func buildSummaryAutogenRoute(
@@ -244,18 +248,20 @@ func buildSummaryAutogenRoute(
248 labels metrix.LabelView,
249 meta metrix.SeriesMeta,
250 policy AutogenPolicy,
251 + typeIDPrefix string,
252 ) (autogenRoute, bool, error) {
253 builder, ok := summaryRoleBuilders[meta.FlattenRole]
254 if !ok {
255 return autogenRoute{}, false, nil
256 }
252 - return builder(metricName, labels, policy)
257 + return builder(metricName, labels, policy, typeIDPrefix)
258 }
259
260 func buildHistogramBucketAutogenRoute(
261 metricName string,
262 labels metrix.LabelView,
263 policy AutogenPolicy,
264 + typeIDPrefix string,
265 ) (autogenRoute, bool, error) {
266 baseName := strings.TrimSuffix(metricName, "_bucket")
267 if baseName == "" {
@@ -268,7 +274,7 @@ func buildHistogramBucketAutogenRoute(
274 chartID := buildJoinedLabelAutogenID(baseName, labels, map[string]struct{}{
275 histogramBucketLabel: {},
276 })
271 - if !fitsTypeIDBudget(policy, chartID) {
277 + if !fitsTypeIDBudget(policy.MaxTypeIDLen, typeIDPrefix, chartID) {
278 return autogenRoute{}, false, nil
279 }
280 return autogenRoute{
@@ -289,30 +295,33 @@ func buildHistogramCountAutogenRoute(
295 metricName string,
296 labels metrix.LabelView,
297 policy AutogenPolicy,
298 + typeIDPrefix string,
299 ) (autogenRoute, bool, error) {
300 baseName := strings.TrimSuffix(metricName, "_count")
301 if baseName == "" {
302 baseName = metricName
303 }
297 - return buildCounterComponentAutogenRoute(baseName, "_count", labels, policy, "events/s")
304 + return buildCounterComponentAutogenRoute(baseName, "_count", labels, policy, typeIDPrefix, "events/s")
305 }
306
307 func buildHistogramSumAutogenRoute(
308 metricName string,
309 labels metrix.LabelView,
310 policy AutogenPolicy,
311 + typeIDPrefix string,
312 ) (autogenRoute, bool, error) {
313 baseName := strings.TrimSuffix(metricName, "_sum")
314 if baseName == "" {
315 baseName = metricName
316 }
309 - return buildCounterComponentAutogenRoute(baseName, "_sum", labels, policy, getAutogenCounterUnits(baseName))
317 + return buildCounterComponentAutogenRoute(baseName, "_sum", labels, policy, typeIDPrefix, getAutogenCounterUnits(baseName))
318 }
319
320 func buildSummaryQuantileAutogenRoute(
321 metricName string,
322 labels metrix.LabelView,
323 policy AutogenPolicy,
324 + typeIDPrefix string,
325 ) (autogenRoute, bool, error) {
326 quantile, ok := labels.Get(summaryQuantileLabel)
327 if !ok || strings.TrimSpace(quantile) == "" {
@@ -321,7 +330,7 @@ func buildSummaryQuantileAutogenRoute(
330 chartID := buildJoinedLabelAutogenID(metricName, labels, map[string]struct{}{
331 summaryQuantileLabel: {},
332 })
324 - if !fitsTypeIDBudget(policy, chartID) {
333 + if !fitsTypeIDBudget(policy.MaxTypeIDLen, typeIDPrefix, chartID) {
334 return autogenRoute{}, false, nil
335 }
336 units := getAutogenSummaryUnits(metricName)
@@ -343,24 +352,26 @@ func buildSummaryCountAutogenRoute(
352 metricName string,
353 labels metrix.LabelView,
354 policy AutogenPolicy,
355 + typeIDPrefix string,
356 ) (autogenRoute, bool, error) {
357 baseName := strings.TrimSuffix(metricName, "_count")
358 if baseName == "" {
359 baseName = metricName
360 }
351 - return buildCounterComponentAutogenRoute(baseName, "_count", labels, policy, "events/s")
361 + return buildCounterComponentAutogenRoute(baseName, "_count", labels, policy, typeIDPrefix, "events/s")
362 }
363
364 func buildSummarySumAutogenRoute(
365 metricName string,
366 labels metrix.LabelView,
367 policy AutogenPolicy,
368 + typeIDPrefix string,
369 ) (autogenRoute, bool, error) {
370 baseName := strings.TrimSuffix(metricName, "_sum")
371 if baseName == "" {
372 baseName = metricName
373 }
363 - return buildCounterComponentAutogenRoute(baseName, "_sum", labels, policy, getAutogenCounterUnits(baseName))
374 + return buildCounterComponentAutogenRoute(baseName, "_sum", labels, policy, typeIDPrefix, getAutogenCounterUnits(baseName))
375 }
376
377 func buildCounterComponentAutogenRoute(
@@ -368,11 +379,12 @@ func buildCounterComponentAutogenRoute(
379 suffix string,
380 labels metrix.LabelView,
381 policy AutogenPolicy,
382 + typeIDPrefix string,
383 units string,
384 ) (autogenRoute, bool, error) {
385 chartName := baseName + suffix
386 chartID := buildJoinedLabelAutogenID(chartName, labels, nil)
375 - if !fitsTypeIDBudget(policy, chartID) {
387 + if !fitsTypeIDBudget(policy.MaxTypeIDLen, typeIDPrefix, chartID) {
388 return autogenRoute{}, false, nil
389 }
390 return autogenRoute{
@@ -393,6 +405,7 @@ func buildStateSetAutogenRoute(
405 labels metrix.LabelView,
406 meta metrix.SeriesMeta,
407 policy AutogenPolicy,
408 + typeIDPrefix string,
409 ) (autogenRoute, bool, error) {
410 if meta.FlattenRole != metrix.FlattenRoleStateSetState {
411 return autogenRoute{}, false, nil
@@ -404,7 +417,7 @@ func buildStateSetAutogenRoute(
417 chartID := buildJoinedLabelAutogenID(metricName, labels, map[string]struct{}{
418 metricName: {},
419 })
407 - if !fitsTypeIDBudget(policy, chartID) {
420 + if !fitsTypeIDBudget(policy.MaxTypeIDLen, typeIDPrefix, chartID) {
421 return autogenRoute{}, false, nil
422 }
423 return autogenRoute{
@@ -426,9 +439,10 @@ func buildScalarAutogenRoute(
439 labels metrix.LabelView,
440 meta metrix.SeriesMeta,
441 policy AutogenPolicy,
442 + typeIDPrefix string,
443 ) (autogenRoute, bool, error) {
444 chartID := buildJoinedLabelAutogenID(metricName, labels, nil)
431 - if !fitsTypeIDBudget(policy, chartID) {
445 + if !fitsTypeIDBudget(policy.MaxTypeIDLen, typeIDPrefix, chartID) {
446 return autogenRoute{}, false, nil
447 }
448 algorithm := program.AlgorithmAbsolute
@@ -450,15 +464,14 @@ func buildScalarAutogenRoute(
464 }, true, nil
465 }
466
453 -func fitsTypeIDBudget(policy AutogenPolicy, chartID string) bool {
454 - maxLen := policy.MaxTypeIDLen
467 +func fitsTypeIDBudget(maxLen int, typeIDPrefix, chartID string) bool {
468 if maxLen <= 0 {
469 maxLen = defaultMaxTypeIDLen
470 }
458 - if strings.TrimSpace(policy.TypeID) == "" {
471 + if strings.TrimSpace(typeIDPrefix) == "" {
472 return len(chartID) <= maxLen
473 }
461 - return len(policy.TypeID)+1+len(chartID) <= maxLen
474 + return len(typeIDPrefix)+1+len(chartID) <= maxLen
475 }
476
477 func buildJoinedLabelAutogenID(metricName string, labels metrix.LabelView, exclude map[string]struct{}) string {
src/go/plugin/framework/chartengine/autogen_test.go
+46 -16
@@ -13,7 +13,30 @@ import (
13 "github.com/netdata/netdata/go/plugins/pkg/metrix"
14 )
15
16 -func TestBuildScalarAutogenRoute(t *testing.T) {
16 +func TestAutogenRouteBuilderScenarios(t *testing.T) {
17 + tests := map[string]struct {
18 + run func(t *testing.T)
19 + }{
20 + "build scalar autogen route": {
21 + run: runTestBuildScalarAutogenRoute,
22 + },
23 + "build histogram bucket autogen route": {
24 + run: runTestBuildHistogramBucketAutogenRoute,
25 + },
26 + "build summary quantile autogen route": {
27 + run: runTestBuildSummaryQuantileAutogenRoute,
28 + },
29 + "build state-set autogen route": {
30 + run: runTestBuildStateSetAutogenRoute,
31 + },
32 + }
33 +
34 + for name, tc := range tests {
35 + t.Run(name, tc.run)
36 + }
37 +}
38 +
39 +func runTestBuildScalarAutogenRoute(t *testing.T) {
40 tests := map[string]struct {
41 metricName string
42 labels map[string]string
@@ -55,6 +78,7 @@ func TestBuildScalarAutogenRoute(t *testing.T) {
78 sortedLabelView(tc.labels),
79 tc.meta,
80 AutogenPolicy{Enabled: true, MaxTypeIDLen: defaultMaxTypeIDLen},
81 + "",
82 )
83 require.NoError(t, err)
84 require.True(t, ok)
@@ -68,7 +92,7 @@ func TestBuildScalarAutogenRoute(t *testing.T) {
92 }
93 }
94
71 -func TestBuildHistogramBucketAutogenRoute(t *testing.T) {
95 +func runTestBuildHistogramBucketAutogenRoute(t *testing.T) {
96 tests := map[string]struct {
97 metricName string
98 labels map[string]string
@@ -93,6 +117,7 @@ func TestBuildHistogramBucketAutogenRoute(t *testing.T) {
117 tc.metricName,
118 sortedLabelView(tc.labels),
119 AutogenPolicy{Enabled: true, MaxTypeIDLen: defaultMaxTypeIDLen},
120 + "",
121 )
122 require.NoError(t, err)
123 require.True(t, ok)
@@ -106,7 +131,7 @@ func TestBuildHistogramBucketAutogenRoute(t *testing.T) {
131 }
132 }
133
109 -func TestBuildSummaryQuantileAutogenRoute(t *testing.T) {
134 +func runTestBuildSummaryQuantileAutogenRoute(t *testing.T) {
135 tests := map[string]struct {
136 metricName string
137 labels map[string]string
@@ -131,6 +156,7 @@ func TestBuildSummaryQuantileAutogenRoute(t *testing.T) {
156 tc.metricName,
157 sortedLabelView(tc.labels),
158 AutogenPolicy{Enabled: true, MaxTypeIDLen: defaultMaxTypeIDLen},
159 + "",
160 )
161 require.NoError(t, err)
162 require.True(t, ok)
@@ -144,7 +170,7 @@ func TestBuildSummaryQuantileAutogenRoute(t *testing.T) {
170 }
171 }
172
147 -func TestBuildStateSetAutogenRoute(t *testing.T) {
173 +func runTestBuildStateSetAutogenRoute(t *testing.T) {
174 tests := map[string]struct {
175 metricName string
176 labels map[string]string
@@ -171,6 +197,7 @@ func TestBuildStateSetAutogenRoute(t *testing.T) {
197 sortedLabelView(tc.labels),
198 tc.meta,
199 AutogenPolicy{Enabled: true, MaxTypeIDLen: defaultMaxTypeIDLen},
200 + "",
201 )
202 require.NoError(t, err)
203 require.True(t, ok)
@@ -187,35 +214,38 @@ func TestBuildStateSetAutogenRoute(t *testing.T) {
214
215 func TestFitsTypeIDBudget(t *testing.T) {
216 tests := map[string]struct {
190 - policy AutogenPolicy
191 - chartID string
192 - want bool
217 + maxLen int
218 + typeIDPrefix string
219 + chartID string
220 + want bool
221 }{
222 "empty type id at exact limit passes": {
195 - policy: AutogenPolicy{MaxTypeIDLen: 5},
223 + maxLen: 5,
224 chartID: "abcde",
225 want: true,
226 },
227 "empty type id over limit fails": {
200 - policy: AutogenPolicy{MaxTypeIDLen: 5},
228 + maxLen: 5,
229 chartID: "abcdef",
230 want: false,
231 },
232 "type id includes separator in budget": {
205 - policy: AutogenPolicy{TypeID: "collector.job", MaxTypeIDLen: 16},
206 - chartID: "abc",
207 - want: false, // len("collector.job")+1+len("abc") == 17 > 16
233 + maxLen: 16,
234 + typeIDPrefix: "collector.job",
235 + chartID: "abc",
236 + want: false, // len("collector.job")+1+len("abc") == 17 > 16
237 },
238 "type id budget overflow fails": {
210 - policy: AutogenPolicy{TypeID: "collector.job", MaxTypeIDLen: 15},
211 - chartID: "abc",
212 - want: false,
239 + maxLen: 15,
240 + typeIDPrefix: "collector.job",
241 + chartID: "abc",
242 + want: false,
243 },
244 }
245
246 for name, tc := range tests {
247 t.Run(name, func(t *testing.T) {
218 - assert.Equal(t, tc.want, fitsTypeIDBudget(tc.policy, tc.chartID))
248 + assert.Equal(t, tc.want, fitsTypeIDBudget(tc.maxLen, tc.typeIDPrefix, tc.chartID))
249 })
250 }
251 }
src/go/plugin/framework/chartengine/build_seq_guard.go
+18
@@ -27,6 +27,24 @@ func (e *Engine) observeBuildSuccessSeq(seq uint64) buildSeqObservation {
27 return obs
28 }
29
30 + if e.state.cfg.runtimePlanner {
31 + if seq < state.lastSuccess {
32 + if !state.violating {
33 + state.violating = true
34 + obs.transition = buildSeqTransitionBroken
35 + }
36 + return obs
37 + }
38 + if state.violating {
39 + state.violating = false
40 + obs.transition = buildSeqTransitionRecovered
41 + }
42 + if seq > state.lastSuccess {
43 + state.lastSuccess = seq
44 + }
45 + return obs
46 + }
47 +
48 if seq <= state.lastSuccess {
49 if !state.violating {
50 state.violating = true
src/go/plugin/framework/chartengine/build_seq_guard_test.go
+60 -20
@@ -10,30 +10,70 @@ import (
10 )
11
12 func TestObserveBuildSuccessSeqTransitions(t *testing.T) {
13 - e, err := New()
14 - require.NoError(t, err)
13 + tests := map[string]struct {
14 + run func(t *testing.T)
15 + }{
16 + "collector mode transitions": {
17 + run: func(t *testing.T) {
18 + e, err := New()
19 + require.NoError(t, err)
20
16 - obs := e.observeBuildSuccessSeq(10)
17 - assert.Equal(t, buildSeqTransitionNone, obs.transition)
18 - assert.Equal(t, uint64(0), obs.previous)
21 + obs := e.observeBuildSuccessSeq(10)
22 + assert.Equal(t, buildSeqTransitionNone, obs.transition)
23 + assert.Equal(t, uint64(0), obs.previous)
24
20 - obs = e.observeBuildSuccessSeq(10)
21 - assert.Equal(t, buildSeqTransitionBroken, obs.transition)
22 - assert.Equal(t, uint64(10), obs.previous)
25 + obs = e.observeBuildSuccessSeq(10)
26 + assert.Equal(t, buildSeqTransitionBroken, obs.transition)
27 + assert.Equal(t, uint64(10), obs.previous)
28
24 - obs = e.observeBuildSuccessSeq(10)
25 - assert.Equal(t, buildSeqTransitionNone, obs.transition)
26 - assert.Equal(t, uint64(10), obs.previous)
29 + obs = e.observeBuildSuccessSeq(10)
30 + assert.Equal(t, buildSeqTransitionNone, obs.transition)
31 + assert.Equal(t, uint64(10), obs.previous)
32
28 - obs = e.observeBuildSuccessSeq(9)
29 - assert.Equal(t, buildSeqTransitionNone, obs.transition)
30 - assert.Equal(t, uint64(10), obs.previous)
33 + obs = e.observeBuildSuccessSeq(9)
34 + assert.Equal(t, buildSeqTransitionNone, obs.transition)
35 + assert.Equal(t, uint64(10), obs.previous)
36
32 - obs = e.observeBuildSuccessSeq(11)
33 - assert.Equal(t, buildSeqTransitionRecovered, obs.transition)
34 - assert.Equal(t, uint64(10), obs.previous)
37 + obs = e.observeBuildSuccessSeq(11)
38 + assert.Equal(t, buildSeqTransitionRecovered, obs.transition)
39 + assert.Equal(t, uint64(10), obs.previous)
40
36 - obs = e.observeBuildSuccessSeq(12)
37 - assert.Equal(t, buildSeqTransitionNone, obs.transition)
38 - assert.Equal(t, uint64(11), obs.previous)
41 + obs = e.observeBuildSuccessSeq(12)
42 + assert.Equal(t, buildSeqTransitionNone, obs.transition)
43 + assert.Equal(t, uint64(11), obs.previous)
44 + },
45 + },
46 + "runtime mode transitions": {
47 + run: func(t *testing.T) {
48 + e, err := New(WithRuntimePlannerMode())
49 + require.NoError(t, err)
50 +
51 + obs := e.observeBuildSuccessSeq(10)
52 + assert.Equal(t, buildSeqTransitionNone, obs.transition)
53 + assert.Equal(t, uint64(0), obs.previous)
54 +
55 + // Stable sequence is expected in runtime mode (no-write ticks).
56 + obs = e.observeBuildSuccessSeq(10)
57 + assert.Equal(t, buildSeqTransitionNone, obs.transition)
58 + assert.Equal(t, uint64(10), obs.previous)
59 +
60 + obs = e.observeBuildSuccessSeq(9)
61 + assert.Equal(t, buildSeqTransitionBroken, obs.transition)
62 + assert.Equal(t, uint64(10), obs.previous)
63 +
64 + // Recovery is allowed once sequence catches up to previous value.
65 + obs = e.observeBuildSuccessSeq(10)
66 + assert.Equal(t, buildSeqTransitionRecovered, obs.transition)
67 + assert.Equal(t, uint64(10), obs.previous)
68 +
69 + obs = e.observeBuildSuccessSeq(10)
70 + assert.Equal(t, buildSeqTransitionNone, obs.transition)
71 + assert.Equal(t, uint64(10), obs.previous)
72 + },
73 + },
74 + }
75 +
76 + for name, tc := range tests {
77 + t.Run(name, tc.run)
78 + }
79 }
src/go/plugin/framework/chartengine/internal/cache/route_cache_test.go
+100 -86
@@ -11,90 +11,104 @@ import (
11 )
12
13 func TestRouteCache(t *testing.T) {
14 - t.Run("stores positive and negative entries", func(t *testing.T) {
15 - rc := NewRouteCache[string]()
16 -
17 - a := metrix.SeriesIdentity{ID: "a", Hash64: 1}
18 - b := metrix.SeriesIdentity{ID: "b", Hash64: 1}
19 -
20 - rc.Store(a, 1, 1, []string{"chart-a"})
21 - rc.Store(b, 1, 1, nil)
22 -
23 - valsA, ok := rc.Lookup(a, 1, 1)
24 - assert.True(t, ok)
25 - assert.Equal(t, []string{"chart-a"}, valsA)
26 -
27 - valsB, ok := rc.Lookup(b, 1, 1)
28 - assert.True(t, ok)
29 - assert.Nil(t, valsB)
30 - })
31 -
32 - t.Run("revision mismatch misses", func(t *testing.T) {
33 - rc := NewRouteCache[string]()
34 - id := metrix.SeriesIdentity{ID: "a", Hash64: 1}
35 - rc.Store(id, 1, 1, []string{"chart-a"})
36 -
37 - _, ok := rc.Lookup(id, 2, 2)
38 - assert.False(t, ok)
39 - })
40 -
41 - t.Run("revision mismatch does not retain stale entry in same build", func(t *testing.T) {
42 - rc := NewRouteCache[string]()
43 - id := metrix.SeriesIdentity{ID: "a", Hash64: 2}
44 - rc.Store(id, 1, 1, []string{"chart-a"})
45 -
46 - _, ok := rc.Lookup(id, 2, 2)
47 - assert.False(t, ok)
48 -
49 - stats := rc.RetainSeen(2)
50 - assert.Equal(t, 1, stats.EntriesBefore)
51 - assert.Equal(t, 0, stats.EntriesAfter)
52 - assert.Equal(t, 1, stats.Pruned)
53 - assert.True(t, stats.FullDrop)
54 - })
55 -
56 - t.Run("retain prunes by seen build sequence", func(t *testing.T) {
57 - rc := NewRouteCache[string]()
58 - a := metrix.SeriesIdentity{ID: "a", Hash64: 10}
59 - b := metrix.SeriesIdentity{ID: "b", Hash64: 11}
60 - c := metrix.SeriesIdentity{ID: "c", Hash64: 12}
61 -
62 - rc.Store(a, 1, 1, []string{"chart-a"})
63 - rc.Store(b, 1, 1, []string{"chart-b"})
64 - rc.Store(c, 1, 1, nil)
65 -
66 - rc.MarkSeenIfPresent(a, 2)
67 - rc.MarkSeenIfPresent(c, 2)
68 - stats := rc.RetainSeen(2)
69 - assert.Equal(t, 3, stats.EntriesBefore)
70 - assert.Equal(t, 2, stats.EntriesAfter)
71 - assert.Equal(t, 1, stats.Pruned)
72 - assert.False(t, stats.FullDrop)
73 -
74 - _, ok := rc.Lookup(a, 1, 2)
75 - assert.True(t, ok)
76 - _, ok = rc.Lookup(b, 1, 2)
77 - assert.False(t, ok)
78 - _, ok = rc.Lookup(c, 1, 2)
79 - assert.True(t, ok)
80 - })
81 -
82 - t.Run("retain full-drops when no entries seen in build", func(t *testing.T) {
83 - rc := NewRouteCache[string]()
84 - a := metrix.SeriesIdentity{ID: "a", Hash64: 20}
85 - b := metrix.SeriesIdentity{ID: "b", Hash64: 21}
86 - rc.Store(a, 1, 1, []string{"chart-a"})
87 - rc.Store(b, 1, 1, []string{"chart-b"})
88 -
89 - stats := rc.RetainSeen(2)
90 - assert.Equal(t, 2, stats.EntriesBefore)
91 - assert.Equal(t, 0, stats.EntriesAfter)
92 - assert.Equal(t, 2, stats.Pruned)
93 - assert.True(t, stats.FullDrop)
94 -
95 - _, ok := rc.Lookup(a, 1, 2)
96 - assert.False(t, ok)
97 - _, ok = rc.Lookup(b, 1, 2)
98 - assert.False(t, ok)
99 - })
14 + tests := map[string]struct {
15 + run func(t *testing.T)
16 + }{
17 + "stores positive and negative entries": {
18 + run: func(t *testing.T) {
19 + rc := NewRouteCache[string]()
20 +
21 + a := metrix.SeriesIdentity{ID: "a", Hash64: 1}
22 + b := metrix.SeriesIdentity{ID: "b", Hash64: 1}
23 +
24 + rc.Store(a, 1, 1, []string{"chart-a"})
25 + rc.Store(b, 1, 1, nil)
26 +
27 + valsA, ok := rc.Lookup(a, 1, 1)
28 + assert.True(t, ok)
29 + assert.Equal(t, []string{"chart-a"}, valsA)
30 +
31 + valsB, ok := rc.Lookup(b, 1, 1)
32 + assert.True(t, ok)
33 + assert.Nil(t, valsB)
34 + },
35 + },
36 + "revision mismatch misses": {
37 + run: func(t *testing.T) {
38 + rc := NewRouteCache[string]()
39 + id := metrix.SeriesIdentity{ID: "a", Hash64: 1}
40 + rc.Store(id, 1, 1, []string{"chart-a"})
41 +
42 + _, ok := rc.Lookup(id, 2, 2)
43 + assert.False(t, ok)
44 + },
45 + },
46 + "revision mismatch does not retain stale entry in same build": {
47 + run: func(t *testing.T) {
48 + rc := NewRouteCache[string]()
49 + id := metrix.SeriesIdentity{ID: "a", Hash64: 2}
50 + rc.Store(id, 1, 1, []string{"chart-a"})
51 +
52 + _, ok := rc.Lookup(id, 2, 2)
53 + assert.False(t, ok)
54 +
55 + stats := rc.RetainSeen(2)
56 + assert.Equal(t, 1, stats.EntriesBefore)
57 + assert.Equal(t, 0, stats.EntriesAfter)
58 + assert.Equal(t, 1, stats.Pruned)
59 + assert.True(t, stats.FullDrop)
60 + },
61 + },
62 + "retain prunes by seen build sequence": {
63 + run: func(t *testing.T) {
64 + rc := NewRouteCache[string]()
65 + a := metrix.SeriesIdentity{ID: "a", Hash64: 10}
66 + b := metrix.SeriesIdentity{ID: "b", Hash64: 11}
67 + c := metrix.SeriesIdentity{ID: "c", Hash64: 12}
68 +
69 + rc.Store(a, 1, 1, []string{"chart-a"})
70 + rc.Store(b, 1, 1, []string{"chart-b"})
71 + rc.Store(c, 1, 1, nil)
72 +
73 + rc.MarkSeenIfPresent(a, 2)
74 + rc.MarkSeenIfPresent(c, 2)
75 + stats := rc.RetainSeen(2)
76 + assert.Equal(t, 3, stats.EntriesBefore)
77 + assert.Equal(t, 2, stats.EntriesAfter)
78 + assert.Equal(t, 1, stats.Pruned)
79 + assert.False(t, stats.FullDrop)
80 +
81 + _, ok := rc.Lookup(a, 1, 2)
82 + assert.True(t, ok)
83 + _, ok = rc.Lookup(b, 1, 2)
84 + assert.False(t, ok)
85 + _, ok = rc.Lookup(c, 1, 2)
86 + assert.True(t, ok)
87 + },
88 + },
89 + "retain full-drops when no entries seen in build": {
90 + run: func(t *testing.T) {
91 + rc := NewRouteCache[string]()
92 + a := metrix.SeriesIdentity{ID: "a", Hash64: 20}
93 + b := metrix.SeriesIdentity{ID: "b", Hash64: 21}
94 + rc.Store(a, 1, 1, []string{"chart-a"})
95 + rc.Store(b, 1, 1, []string{"chart-b"})
96 +
97 + stats := rc.RetainSeen(2)
98 + assert.Equal(t, 2, stats.EntriesBefore)
99 + assert.Equal(t, 0, stats.EntriesAfter)
100 + assert.Equal(t, 2, stats.Pruned)
101 + assert.True(t, stats.FullDrop)
102 +
103 + _, ok := rc.Lookup(a, 1, 2)
104 + assert.False(t, ok)
105 + _, ok = rc.Lookup(b, 1, 2)
106 + assert.False(t, ok)
107 + },
108 + },
109 + }
110 +
111 + for name, tc := range tests {
112 + t.Run(name, tc.run)
113 + }
114 }
src/go/plugin/framework/chartengine/lifecycle_test.go
+40 -34
@@ -10,41 +10,47 @@ import (
10 )
11
12 func TestEnforceChartInstanceCapsSoftWhenAllExistingAreActive(t *testing.T) {
13 - const currentSuccessSeq = 42
14 - const templateID = "g0.c0"
13 + tests := map[string]struct{}{"all existing active instances are kept": {}}
14
16 - lifecycle := program.LifecyclePolicy{
17 - MaxInstances: 1,
18 - }
19 - state := newMaterializedState()
20 - state.charts["win_nic_traffic_eth0"] = &materializedChartState{
21 - templateID: templateID,
22 - lifecycle: lifecycle,
23 - lastSeenSuccessSeq: currentSuccessSeq,
24 - dimensions: make(map[string]*materializedDimensionState),
25 - }
26 - state.charts["win_nic_traffic_eth1"] = &materializedChartState{
27 - templateID: templateID,
28 - lifecycle: lifecycle,
29 - lastSeenSuccessSeq: currentSuccessSeq,
30 - dimensions: make(map[string]*materializedDimensionState),
31 - }
15 + for name := range tests {
16 + t.Run(name, func(t *testing.T) {
17 + const currentSuccessSeq = 42
18 + const templateID = "g0.c0"
19
33 - chartsByID := map[string]*chartState{
34 - "win_nic_traffic_eth0": {
35 - templateID: templateID,
36 - lifecycle: lifecycle,
37 - entries: make(map[string]*dimBuildEntry),
38 - },
39 - "win_nic_traffic_eth1": {
40 - templateID: templateID,
41 - lifecycle: lifecycle,
42 - entries: make(map[string]*dimBuildEntry),
43 - },
44 - }
20 + lifecycle := program.LifecyclePolicy{
21 + MaxInstances: 1,
22 + }
23 + state := newMaterializedState()
24 + state.charts["win_nic_traffic_eth0"] = &materializedChartState{
25 + templateID: templateID,
26 + lifecycle: lifecycle,
27 + lastSeenSuccessSeq: currentSuccessSeq,
28 + dimensions: make(map[string]*materializedDimensionState),
29 + }
30 + state.charts["win_nic_traffic_eth1"] = &materializedChartState{
31 + templateID: templateID,
32 + lifecycle: lifecycle,
33 + lastSeenSuccessSeq: currentSuccessSeq,
34 + dimensions: make(map[string]*materializedDimensionState),
35 + }
36 +
37 + chartsByID := map[string]*chartState{
38 + "win_nic_traffic_eth0": {
39 + templateID: templateID,
40 + lifecycle: lifecycle,
41 + entries: make(map[string]*dimBuildEntry),
42 + },
43 + "win_nic_traffic_eth1": {
44 + templateID: templateID,
45 + lifecycle: lifecycle,
46 + entries: make(map[string]*dimBuildEntry),
47 + },
48 + }
49
46 - removeCharts := enforceChartInstanceCaps(currentSuccessSeq, chartsByID, &state)
47 - assert.Empty(t, removeCharts)
48 - assert.Len(t, chartsByID, 2)
49 - assert.Len(t, state.charts, 2)
50 + removeCharts := enforceChartInstanceCaps(currentSuccessSeq, chartsByID, &state)
51 + assert.Empty(t, removeCharts)
52 + assert.Len(t, chartsByID, 2)
53 + assert.Len(t, state.charts, 2)
54 + })
55 + }
56 }
src/go/plugin/framework/chartengine/matcher_test.go
+65 -51
@@ -10,55 +10,69 @@ import (
10 "github.com/netdata/netdata/go/plugins/pkg/metrix"
11 )
12
13 -func TestRouteCacheStoresPositiveAndNegativeRoutes(t *testing.T) {
14 - cache := newRouteCache()
15 -
16 - a := metrix.SeriesIdentity{ID: "a", Hash64: 1}
17 - b := metrix.SeriesIdentity{ID: "b", Hash64: 1}
18 -
19 - cache.Store(a, 1, 1, []routeBinding{{ChartID: "ca"}})
20 - cache.Store(b, 1, 1, nil) // negative-cache entry
21 -
22 - routesA, ok := cache.Lookup(a, 1, 1)
23 - assert.True(t, ok)
24 - assert.Equal(t, "ca", routesA[0].ChartID)
25 -
26 - routesB, ok := cache.Lookup(b, 1, 1)
27 - assert.True(t, ok)
28 - assert.Empty(t, routesB)
29 -}
30 -
31 -func TestRouteCacheRetainSeenPrunesByBuildSequence(t *testing.T) {
32 - cache := newRouteCache()
33 -
34 - a := metrix.SeriesIdentity{ID: "a", Hash64: 10}
35 - b := metrix.SeriesIdentity{ID: "b", Hash64: 11}
36 - c := metrix.SeriesIdentity{ID: "c", Hash64: 12}
37 -
38 - cache.Store(a, 1, 1, []routeBinding{{ChartID: "ca"}})
39 - cache.Store(b, 1, 1, []routeBinding{{ChartID: "cb"}})
40 - cache.Store(c, 1, 1, nil)
41 -
42 - cache.MarkSeenIfPresent(a, 2)
43 - cache.MarkSeenIfPresent(c, 2)
44 - cache.RetainSeen(2)
45 -
46 - _, ok := cache.Lookup(a, 1, 2)
47 - assert.True(t, ok)
48 -
49 - _, ok = cache.Lookup(b, 1, 2)
50 - assert.False(t, ok)
51 -
52 - _, ok = cache.Lookup(c, 1, 2)
53 - assert.True(t, ok)
54 -}
55 -
56 -func TestRouteCacheLookupMissOnRevisionChange(t *testing.T) {
57 - cache := newRouteCache()
58 - id := metrix.SeriesIdentity{ID: "a", Hash64: 1}
59 -
60 - cache.Store(id, 1, 1, []routeBinding{{ChartID: "ca"}})
61 -
62 - _, ok := cache.Lookup(id, 2, 2)
63 - assert.False(t, ok)
13 +func TestRouteCacheScenarios(t *testing.T) {
14 + tests := map[string]struct {
15 + run func(t *testing.T)
16 + }{
17 + "stores positive and negative routes": {
18 + run: func(t *testing.T) {
19 + cache := newRouteCache()
20 +
21 + a := metrix.SeriesIdentity{ID: "a", Hash64: 1}
22 + b := metrix.SeriesIdentity{ID: "b", Hash64: 1}
23 +
24 + cache.Store(a, 1, 1, []routeBinding{{ChartID: "ca"}})
25 + cache.Store(b, 1, 1, nil) // negative-cache entry
26 +
27 + routesA, ok := cache.Lookup(a, 1, 1)
28 + assert.True(t, ok)
29 + assert.Equal(t, "ca", routesA[0].ChartID)
30 +
31 + routesB, ok := cache.Lookup(b, 1, 1)
32 + assert.True(t, ok)
33 + assert.Empty(t, routesB)
34 + },
35 + },
36 + "retain seen prunes by build sequence": {
37 + run: func(t *testing.T) {
38 + cache := newRouteCache()
39 +
40 + a := metrix.SeriesIdentity{ID: "a", Hash64: 10}
41 + b := metrix.SeriesIdentity{ID: "b", Hash64: 11}
42 + c := metrix.SeriesIdentity{ID: "c", Hash64: 12}
43 +
44 + cache.Store(a, 1, 1, []routeBinding{{ChartID: "ca"}})
45 + cache.Store(b, 1, 1, []routeBinding{{ChartID: "cb"}})
46 + cache.Store(c, 1, 1, nil)
47 +
48 + cache.MarkSeenIfPresent(a, 2)
49 + cache.MarkSeenIfPresent(c, 2)
50 + cache.RetainSeen(2)
51 +
52 + _, ok := cache.Lookup(a, 1, 2)
53 + assert.True(t, ok)
54 +
55 + _, ok = cache.Lookup(b, 1, 2)
56 + assert.False(t, ok)
57 +
58 + _, ok = cache.Lookup(c, 1, 2)
59 + assert.True(t, ok)
60 + },
61 + },
62 + "lookup misses on revision change": {
63 + run: func(t *testing.T) {
64 + cache := newRouteCache()
65 + id := metrix.SeriesIdentity{ID: "a", Hash64: 1}
66 +
67 + cache.Store(id, 1, 1, []routeBinding{{ChartID: "ca"}})
68 +
69 + _, ok := cache.Lookup(id, 2, 2)
70 + assert.False(t, ok)
71 + },
72 + },
73 + }
74 +
75 + for name, tc := range tests {
76 + t.Run(name, tc.run)
77 + }
78 }
src/go/plugin/framework/chartengine/options.go
+39 -28
@@ -13,6 +13,7 @@ import (
13
14 type engineConfig struct {
15 autogen AutogenPolicy
16 + autogenTypeID string
17 selector metrixselector.Selector
18 autogenOverride policyOverride[AutogenPolicy]
19 selectorOverride policyOverride[metrixselector.Selector]
@@ -20,6 +21,7 @@ type engineConfig struct {
21 runtimeStoreSet bool
22 log *logger.Logger
23 seriesSelection seriesSelectionMode
24 + runtimePlanner bool
25 }
26
27 type policyOverride[T any] struct {
@@ -48,11 +50,11 @@ type AutogenPolicy = runtimecomp.AutogenPolicy
50 // EnginePolicy controls chartengine matching/materialization behavior.
51 type EnginePolicy struct {
52 // Selector filters input series globally before template/autogen routing.
51 - // Nil or empty selector means "allow all".
53 + // Nil means "no override", and an explicitly empty expr means "override to allow all".
54 Selector *metrixselector.Expr
55
56 // Autogen controls unmatched-series fallback behavior.
55 - Autogen AutogenPolicy
57 + Autogen *AutogenPolicy
58 }
59
60 func defaultAutogenPolicy() AutogenPolicy {
@@ -74,8 +76,8 @@ func normalizeAutogenPolicy(policy AutogenPolicy) (AutogenPolicy, error) {
76 return policy, nil
77 }
78
77 -func compileEngineSelector(expr *metrixselector.Expr) (metrixselector.Selector, error) {
78 - if expr == nil || expr.Empty() {
79 +func compileEngineSelector(expr metrixselector.Expr) (metrixselector.Selector, error) {
80 + if expr.Empty() {
81 return nil, nil
82 }
83 return expr.Parse()
@@ -84,32 +86,22 @@ func compileEngineSelector(expr *metrixselector.Expr) (metrixselector.Selector,
86 // WithEnginePolicy configures chartengine matching/materialization policy.
87 func WithEnginePolicy(policy EnginePolicy) Option {
88 return func(cfg *engineConfig) error {
87 - autogen, err := normalizeAutogenPolicy(policy.Autogen)
88 - if err != nil {
89 - return err
89 + if policy.Autogen != nil {
90 + autogen, err := normalizeAutogenPolicy(*policy.Autogen)
91 + if err != nil {
92 + return err
93 + }
94 + cfg.autogenOverride = policyOverride[AutogenPolicy]{set: true, value: autogen}
95 + cfg.autogen = autogen
96 }
91 - selector, err := compileEngineSelector(policy.Selector)
92 - if err != nil {
93 - return fmt.Errorf("invalid engine selector: %w", err)
97 + if policy.Selector != nil {
98 + selector, err := compileEngineSelector(*policy.Selector)
99 + if err != nil {
100 + return fmt.Errorf("invalid engine selector: %w", err)
101 + }
102 + cfg.selectorOverride = policyOverride[metrixselector.Selector]{set: true, value: selector}
103 + cfg.selector = selector
104 }
95 - cfg.autogenOverride = policyOverride[AutogenPolicy]{set: true, value: autogen}
96 - cfg.selectorOverride = policyOverride[metrixselector.Selector]{set: true, value: selector}
97 - cfg.autogen = autogen
98 - cfg.selector = selector
99 - return nil
100 - }
101 -}
102 -
103 -// WithAutogenPolicy configures unmatched-series autogen behavior.
104 -// Deprecated: prefer WithEnginePolicy.
105 -func WithAutogenPolicy(policy AutogenPolicy) Option {
106 - return func(cfg *engineConfig) error {
107 - autogen, err := normalizeAutogenPolicy(policy)
108 - if err != nil {
109 - return err
110 - }
111 - cfg.autogenOverride = policyOverride[AutogenPolicy]{set: true, value: autogen}
112 - cfg.autogen = autogen
105 return nil
106 }
107 }
@@ -142,6 +134,25 @@ func WithSeriesSelectionAllVisible() Option {
134 }
135 }
136
137 +// WithRuntimePlannerMode enables runtime/internal planner semantics for
138 +// build-cycle dedupe bookkeeping while keeping lifecycle/cache tied to
139 +// source LastSuccessSeq.
140 +func WithRuntimePlannerMode() Option {
141 + return func(cfg *engineConfig) error {
142 + cfg.runtimePlanner = true
143 + return nil
144 + }
145 +}
146 +
147 +// WithEmitTypeIDBudgetPrefix configures chartengine autogen type-id budget
148 +// checks to use the effective emission type-id prefix (for example job fullName).
149 +func WithEmitTypeIDBudgetPrefix(typeID string) Option {
150 + return func(cfg *engineConfig) error {
151 + cfg.autogenTypeID = typeID
152 + return nil
153 + }
154 +}
155 +
156 func applyOptions(opts ...Option) (engineConfig, error) {
157 cfg := engineConfig{
158 autogen: defaultAutogenPolicy(),
src/go/plugin/framework/chartengine/planner.go
+18 -2
@@ -101,6 +101,7 @@ type planBuildContext struct {
101 out *Plan
102 reader metrix.Reader
103 collectMeta metrix.CollectMeta
104 + buildCycle uint64
105 prog *program.Program
106 cache *routeCache
107 index matchIndex
@@ -151,6 +152,7 @@ func (e *Engine) BuildPlan(reader metrix.Reader) (Plan, error) {
152 defer e.mu.Unlock()
153
154 obs := e.observeBuildSuccessSeq(collectMeta.LastSuccessSeq)
155 + buildCycle := e.nextBuildCycle(collectMeta.LastSuccessSeq)
156 sample.buildSeqViolation = e.state.buildSeq.violating
157 sample.buildSeqObserved = true
158 switch obs.transition {
@@ -171,7 +173,7 @@ func (e *Engine) BuildPlan(reader metrix.Reader) (Plan, error) {
173 }
174
175 phaseStartedAt := time.Now()
174 - ctx, err := e.preparePlanBuildContext(reader, &out, collectMeta)
176 + ctx, err := e.preparePlanBuildContext(reader, &out, collectMeta, buildCycle)
177 sample.phasePrepareSeconds = time.Since(phaseStartedAt).Seconds()
178 if err != nil {
179 sample.buildErr = true
@@ -295,6 +297,7 @@ func (e *Engine) preparePlanBuildContext(
297 reader metrix.Reader,
298 out *Plan,
299 collectMeta metrix.CollectMeta,
300 + buildCycle uint64,
301 ) (*planBuildContext, error) {
302 prog := e.state.program
303 if prog == nil {
@@ -330,6 +333,7 @@ func (e *Engine) preparePlanBuildContext(
333 out: out,
334 reader: reader,
335 collectMeta: collectMeta,
336 + buildCycle: buildCycle,
337 prog: prog,
338 cache: cache,
339 index: index,
@@ -486,7 +490,7 @@ func (ctx *planBuildContext) accumulateRoute(
490 lifecycle: route.Lifecycle,
491 labels: labelsAcc,
492 entries: entries,
489 - currentBuildSeq: ctx.collectMeta.LastSuccessSeq,
493 + currentBuildSeq: ctx.buildCycle,
494 }
495 ctx.chartsByID[route.ChartID] = cs
496 }
@@ -672,3 +676,15 @@ func buildPlan(engine *Engine, reader metrix.Reader) (Plan, error) {
676 func isAutogenTemplateID(templateID string) bool {
677 return strings.HasPrefix(templateID, autogenTemplatePrefix)
678 }
679 +
680 +func (e *Engine) nextBuildCycle(sourceSuccessSeq uint64) uint64 {
681 + if !e.state.cfg.runtimePlanner {
682 + return sourceSuccessSeq
683 + }
684 + e.state.plannerBuildSeq++
685 + // Seen-seq zero value is reserved for "never seen".
686 + if e.state.plannerBuildSeq == 0 {
687 + e.state.plannerBuildSeq = 1
688 + }
689 + return e.state.plannerBuildSeq
690 +}
src/go/plugin/framework/chartengine/planner_lifecycle_test.go
+19 -2
@@ -10,7 +10,24 @@ import (
10 "github.com/stretchr/testify/require"
11 )
12
13 -func TestEnforceLifecycleCaps_DimensionCapEvictsLRU(t *testing.T) {
13 +func TestPlannerLifecycleScenarios(t *testing.T) {
14 + tests := map[string]struct {
15 + run func(t *testing.T)
16 + }{
17 + "enforce lifecycle caps dimension cap evicts lru": {
18 + run: runTestEnforceLifecycleCapsDimensionCapEvictsLRU,
19 + },
20 + "collect expiry removals dimension and chart expiry": {
21 + run: runTestCollectExpiryRemovalsDimensionAndChartExpiry,
22 + },
23 + }
24 +
25 + for name, tc := range tests {
26 + t.Run(name, tc.run)
27 + }
28 +}
29 +
30 +func runTestEnforceLifecycleCapsDimensionCapEvictsLRU(t *testing.T) {
31 tests := map[string]struct {
32 currentSeq uint64
33 }{
@@ -83,7 +100,7 @@ func TestEnforceLifecycleCaps_DimensionCapEvictsLRU(t *testing.T) {
100 }
101 }
102
86 -func TestCollectExpiryRemovals_DimensionAndChartExpiry(t *testing.T) {
103 +func runTestCollectExpiryRemovalsDimensionAndChartExpiry(t *testing.T) {
104 tests := map[string]struct {
105 currentSeq uint64
106 }{
src/go/plugin/framework/chartengine/planner_test.go
+200 -52
@@ -197,7 +197,47 @@ groups:
197 }
198 }
199
200 -func TestBuildPlanRequiresFlattenedReaderForInference(t *testing.T) {
200 +func TestBuildPlanLegacySingleScenarioCases(t *testing.T) {
201 + tests := map[string]struct {
202 + run func(t *testing.T)
203 + }{
204 + "BuildPlanRequiresFlattenedReaderForInference": {run: runTestBuildPlanRequiresFlattenedReaderForInference},
205 + "BuildPlanUsesRouteCacheReuse": {run: runTestBuildPlanUsesRouteCacheReuse},
206 + "BuildPlanLifecycleDimensionExpiry": {run: runTestBuildPlanLifecycleDimensionExpiry},
207 + "BuildPlanLifecycleChartExpiry": {run: runTestBuildPlanLifecycleChartExpiry},
208 + "BuildPlanLifecycleNoRemovalOnFailedCycle": {run: runTestBuildPlanLifecycleNoRemovalOnFailedCycle},
209 + "BuildPlanRendersChartIDsFromInstances": {run: runTestBuildPlanRendersChartIDsFromInstances},
210 + "BuildPlanEnforcesMaxInstancesDeterministically": {run: runTestBuildPlanEnforcesMaxInstancesDeterministically},
211 + "BuildPlanEnforcesMaxDimsDeterministically": {run: runTestBuildPlanEnforcesMaxDimsDeterministically},
212 + "BuildPlanComputesChartLabelsIntersectionAndExclusions": {run: runTestBuildPlanComputesChartLabelsIntersectionAndExclusions},
213 + "BuildPlanAutogenDisabledSkipsUnmatchedSeries": {run: runTestBuildPlanAutogenDisabledSkipsUnmatchedSeries},
214 + "BuildPlanEnginePolicySelectorFiltersSeriesBeforeRouting": {run: runTestBuildPlanEnginePolicySelectorFiltersSeriesBeforeRouting},
215 + "BuildPlanTemplateEnginePolicyControlsSelectorAndAutogen": {run: runTestBuildPlanTemplateEnginePolicyControlsSelectorAndAutogen},
216 + "BuildPlanEnginePolicyOptionOverridesTemplatePolicy": {run: runTestBuildPlanEnginePolicyOptionOverridesTemplatePolicy},
217 + "BuildPlanAutogenOptionKeepsTemplateSelector": {run: runTestBuildPlanAutogenOptionKeepsTemplateSelector},
218 + "BuildPlanAutogenCreatesChartForUnmatchedScalar": {run: runTestBuildPlanAutogenCreatesChartForUnmatchedScalar},
219 + "BuildPlanAutogenUsesMetricMetadataForScalar": {run: runTestBuildPlanAutogenUsesMetricMetadataForScalar},
220 + "BuildPlanAutogenUsesMetricMetadataForHistogram": {run: runTestBuildPlanAutogenUsesMetricMetadataForHistogram},
221 + "BuildPlanAutogenUsesMetricFloatMetadataForScalar": {run: runTestBuildPlanAutogenUsesMetricFloatMetadataForScalar},
222 + "BuildPlanAutogenUsesMetricMetadataForSummaryWithoutQuantiles": {run: runTestBuildPlanAutogenUsesMetricMetadataForSummaryWithoutQuantiles},
223 + "BuildPlanTemplatePrecedenceOverAutogen": {run: runTestBuildPlanTemplatePrecedenceOverAutogen},
224 + "BuildPlanAutogenStrictOverflowDrop": {run: runTestBuildPlanAutogenStrictOverflowDrop},
225 + "BuildPlanAutogenUsesFlattenMetadataForHistogramBuckets": {run: runTestBuildPlanAutogenUsesFlattenMetadataForHistogramBuckets},
226 + "BuildPlanAutogenCreatesChartForUnmatchedGauge": {run: runTestBuildPlanAutogenCreatesChartForUnmatchedGauge},
227 + "BuildPlanAutogenCreatesChartForUnmatchedStateSet": {run: runTestBuildPlanAutogenCreatesChartForUnmatchedStateSet},
228 + "BuildPlanAutogenKeepsStateSetUnitsWhenMetricMetaUnitIsSet": {run: runTestBuildPlanAutogenKeepsStateSetUnitsWhenMetricMetaUnitIsSet},
229 + "BuildPlanTemplateWinsOnAutogenChartIDCollisionAcrossSeries": {run: runTestBuildPlanTemplateWinsOnAutogenChartIDCollisionAcrossSeries},
230 + "BuildPlanAutogenRemovalLifecycleExpiry": {run: runTestBuildPlanAutogenRemovalLifecycleExpiry},
231 + "BuildPlanFirstWriterWinsAndAccumulatesRepeatedRoutes": {run: runTestBuildPlanFirstWriterWinsAndAccumulatesRepeatedRoutes},
232 + "BuildPlanEmptyEmissionAndScratchReusePruneAcrossCycles": {run: runTestBuildPlanEmptyEmissionAndScratchReusePruneAcrossCycles},
233 + }
234 +
235 + for name, tc := range tests {
236 + t.Run(name, tc.run)
237 + }
238 +}
239 +
240 +func runTestBuildPlanRequiresFlattenedReaderForInference(t *testing.T) {
241 e, err := New()
242 require.NoError(t, err)
243
@@ -236,7 +276,7 @@ groups:
276 require.NoError(t, err)
277 }
278
239 -func TestBuildPlanUsesRouteCacheReuse(t *testing.T) {
279 +func runTestBuildPlanUsesRouteCacheReuse(t *testing.T) {
280 e, err := New()
281 require.NoError(t, err)
282
@@ -287,7 +327,7 @@ groups:
327 assert.Equal(t, float64(20), findUpdateAction(plan2).Values[0].Float64)
328 }
329
290 -func TestBuildPlanLifecycleDimensionExpiry(t *testing.T) {
330 +func runTestBuildPlanLifecycleDimensionExpiry(t *testing.T) {
331 e, err := New()
332 require.NoError(t, err)
333
@@ -341,7 +381,7 @@ groups:
381 assert.Equal(t, "ok", removeDim.Name)
382 }
383
344 -func TestBuildPlanLifecycleChartExpiry(t *testing.T) {
384 +func runTestBuildPlanLifecycleChartExpiry(t *testing.T) {
385 e, err := New()
386 require.NoError(t, err)
387
@@ -383,7 +423,7 @@ groups:
423 assert.Equal(t, []ActionKind{ActionRemoveChart}, actionKinds(plan2.Actions))
424 }
425
386 -func TestBuildPlanLifecycleNoRemovalOnFailedCycle(t *testing.T) {
426 +func runTestBuildPlanLifecycleNoRemovalOnFailedCycle(t *testing.T) {
427 e, err := New()
428 require.NoError(t, err)
429
@@ -432,7 +472,7 @@ groups:
472 assert.Equal(t, []ActionKind{ActionRemoveChart}, actionKinds(plan3.Actions))
473 }
474
435 -func TestBuildPlanRendersChartIDsFromInstances(t *testing.T) {
475 +func runTestBuildPlanRendersChartIDsFromInstances(t *testing.T) {
476 e, err := New()
477 require.NoError(t, err)
478
@@ -502,7 +542,7 @@ groups:
542 assert.Equal(t, []ActionKind{ActionUpdateChart, ActionUpdateChart}, actionKinds(plan2.Actions))
543 }
544
505 -func TestBuildPlanEnforcesMaxInstancesDeterministically(t *testing.T) {
545 +func runTestBuildPlanEnforcesMaxInstancesDeterministically(t *testing.T) {
546 e, err := New()
547 require.NoError(t, err)
548
@@ -570,7 +610,7 @@ groups:
610 }, actionKinds(plan3.Actions))
611 }
612
573 -func TestBuildPlanEnforcesMaxDimsDeterministically(t *testing.T) {
613 +func runTestBuildPlanEnforcesMaxDimsDeterministically(t *testing.T) {
614 e, err := New()
615 require.NoError(t, err)
616
@@ -640,7 +680,7 @@ groups:
680 }, actionKinds(plan3.Actions))
681 }
682
643 -func TestBuildPlanComputesChartLabelsIntersectionAndExclusions(t *testing.T) {
683 +func runTestBuildPlanComputesChartLabelsIntersectionAndExclusions(t *testing.T) {
684 e, err := New()
685 require.NoError(t, err)
686
@@ -702,7 +742,7 @@ groups:
742 assert.False(t, hasDirection)
743 }
744
705 -func TestBuildPlanAutogenDisabledSkipsUnmatchedSeries(t *testing.T) {
745 +func runTestBuildPlanAutogenDisabledSkipsUnmatchedSeries(t *testing.T) {
746 e, err := New()
747 require.NoError(t, err)
748
@@ -735,13 +775,13 @@ groups:
775 assert.Empty(t, plan.Actions)
776 }
777
738 -func TestBuildPlanEnginePolicySelectorFiltersSeriesBeforeRouting(t *testing.T) {
778 +func runTestBuildPlanEnginePolicySelectorFiltersSeriesBeforeRouting(t *testing.T) {
779 selectorExpr := metrixselector.Expr{
780 Allow: []string{`svc.errors_total{method="GET"}`},
781 }
782 e, err := New(WithEnginePolicy(EnginePolicy{
783 Selector: &selectorExpr,
744 - Autogen: AutogenPolicy{Enabled: true},
784 + Autogen: &AutogenPolicy{Enabled: true},
785 }))
786 require.NoError(t, err)
787
@@ -788,7 +828,7 @@ groups:
828 assert.Equal(t, float64(10), update.Values[0].Float64)
829 }
830
791 -func TestBuildPlanTemplateEnginePolicyControlsSelectorAndAutogen(t *testing.T) {
831 +func runTestBuildPlanTemplateEnginePolicyControlsSelectorAndAutogen(t *testing.T) {
832 e, err := New()
833 require.NoError(t, err)
834
@@ -826,13 +866,13 @@ groups:
866 assert.Equal(t, "svc.errors_total-method=GET", create.ChartID)
867 }
868
829 -func TestBuildPlanEnginePolicyOptionOverridesTemplatePolicy(t *testing.T) {
869 +func runTestBuildPlanEnginePolicyOptionOverridesTemplatePolicy(t *testing.T) {
870 overrideSelector := metrixselector.Expr{
871 Allow: []string{`svc.errors_total{method="POST"}`},
872 }
873 e, err := New(WithEnginePolicy(EnginePolicy{
874 Selector: &overrideSelector,
835 - Autogen: AutogenPolicy{Enabled: true},
875 + Autogen: &AutogenPolicy{Enabled: true},
876 }))
877 require.NoError(t, err)
878
@@ -870,8 +910,8 @@ groups:
910 assert.Equal(t, "svc.errors_total-method=POST", create.ChartID)
911 }
912
873 -func TestBuildPlanAutogenOptionKeepsTemplateSelector(t *testing.T) {
874 - e, err := New(WithAutogenPolicy(AutogenPolicy{Enabled: true}))
913 +func runTestBuildPlanAutogenOptionKeepsTemplateSelector(t *testing.T) {
914 + e, err := New(WithEnginePolicy(EnginePolicy{Autogen: &AutogenPolicy{Enabled: true}}))
915 require.NoError(t, err)
916
917 yaml := `
@@ -908,8 +948,8 @@ groups:
948 assert.Equal(t, "svc.errors_total-method=GET", create.ChartID)
949 }
950
911 -func TestBuildPlanAutogenCreatesChartForUnmatchedScalar(t *testing.T) {
912 - e, err := New(WithAutogenPolicy(AutogenPolicy{Enabled: true}))
951 +func runTestBuildPlanAutogenCreatesChartForUnmatchedScalar(t *testing.T) {
952 + e, err := New(WithEnginePolicy(EnginePolicy{Autogen: &AutogenPolicy{Enabled: true}}))
953 require.NoError(t, err)
954
955 yaml := `
@@ -956,8 +996,8 @@ groups:
996 assert.Equal(t, float64(10), update.Values[0].Float64)
997 }
998
959 -func TestBuildPlanAutogenUsesMetricMetadataForScalar(t *testing.T) {
960 - e, err := New(WithAutogenPolicy(AutogenPolicy{Enabled: true}))
999 +func runTestBuildPlanAutogenUsesMetricMetadataForScalar(t *testing.T) {
1000 + e, err := New(WithEnginePolicy(EnginePolicy{Autogen: &AutogenPolicy{Enabled: true}}))
1001 require.NoError(t, err)
1002
1003 yaml := `
@@ -999,8 +1039,8 @@ groups:
1039 assert.Equal(t, "bytes/s", create.Meta.Units)
1040 }
1041
1002 -func TestBuildPlanAutogenUsesMetricMetadataForHistogram(t *testing.T) {
1003 - e, err := New(WithAutogenPolicy(AutogenPolicy{Enabled: true}))
1042 +func runTestBuildPlanAutogenUsesMetricMetadataForHistogram(t *testing.T) {
1043 + e, err := New(WithEnginePolicy(EnginePolicy{Autogen: &AutogenPolicy{Enabled: true}}))
1044 require.NoError(t, err)
1045
1046 yaml := `
@@ -1056,8 +1096,8 @@ groups:
1096 assert.Equal(t, "ms/s", sum.Meta.Units)
1097 }
1098
1059 -func TestBuildPlanAutogenUsesMetricFloatMetadataForScalar(t *testing.T) {
1060 - e, err := New(WithAutogenPolicy(AutogenPolicy{Enabled: true}))
1099 +func runTestBuildPlanAutogenUsesMetricFloatMetadataForScalar(t *testing.T) {
1100 + e, err := New(WithEnginePolicy(EnginePolicy{Autogen: &AutogenPolicy{Enabled: true}}))
1101 require.NoError(t, err)
1102
1103 yaml := `
@@ -1108,8 +1148,8 @@ groups:
1148 assert.Equal(t, float64(10.5), update.Values[0].Float64)
1149 }
1150
1111 -func TestBuildPlanAutogenUsesMetricMetadataForSummaryWithoutQuantiles(t *testing.T) {
1112 - e, err := New(WithAutogenPolicy(AutogenPolicy{Enabled: true}))
1151 +func runTestBuildPlanAutogenUsesMetricMetadataForSummaryWithoutQuantiles(t *testing.T) {
1152 + e, err := New(WithEnginePolicy(EnginePolicy{Autogen: &AutogenPolicy{Enabled: true}}))
1153 require.NoError(t, err)
1154
1155 yaml := `
@@ -1154,8 +1194,8 @@ groups:
1194 assert.Equal(t, "ms/s", sum.Meta.Units)
1195 }
1196
1157 -func TestBuildPlanTemplatePrecedenceOverAutogen(t *testing.T) {
1158 - e, err := New(WithAutogenPolicy(AutogenPolicy{Enabled: true}))
1197 +func runTestBuildPlanTemplatePrecedenceOverAutogen(t *testing.T) {
1198 + e, err := New(WithEnginePolicy(EnginePolicy{Autogen: &AutogenPolicy{Enabled: true}}))
1199 require.NoError(t, err)
1200
1201 yaml := `
@@ -1210,12 +1250,14 @@ groups:
1250 assert.Equal(t, int64(10), update.Values[0].Int64)
1251 }
1252
1213 -func TestBuildPlanAutogenStrictOverflowDrop(t *testing.T) {
1214 - e, err := New(WithAutogenPolicy(AutogenPolicy{
1215 - Enabled: true,
1216 - TypeID: "collector.job",
1217 - MaxTypeIDLen: 32,
1218 - }))
1253 +func runTestBuildPlanAutogenStrictOverflowDrop(t *testing.T) {
1254 + e, err := New(
1255 + WithEmitTypeIDBudgetPrefix("collector.job"),
1256 + WithEnginePolicy(EnginePolicy{Autogen: &AutogenPolicy{
1257 + Enabled: true,
1258 + MaxTypeIDLen: 32,
1259 + }}),
1260 + )
1261 require.NoError(t, err)
1262
1263 yaml := `
@@ -1249,8 +1291,8 @@ groups:
1291 assert.Empty(t, plan.Actions)
1292 }
1293
1252 -func TestBuildPlanAutogenUsesFlattenMetadataForHistogramBuckets(t *testing.T) {
1253 - e, err := New(WithAutogenPolicy(AutogenPolicy{Enabled: true}))
1294 +func runTestBuildPlanAutogenUsesFlattenMetadataForHistogramBuckets(t *testing.T) {
1295 + e, err := New(WithEnginePolicy(EnginePolicy{Autogen: &AutogenPolicy{Enabled: true}}))
1296 require.NoError(t, err)
1297
1298 yaml := `
@@ -1318,8 +1360,8 @@ groups:
1360 assert.Contains(t, dims, "bucket_+Inf")
1361 }
1362
1321 -func TestBuildPlanAutogenCreatesChartForUnmatchedGauge(t *testing.T) {
1322 - e, err := New(WithAutogenPolicy(AutogenPolicy{Enabled: true}))
1363 +func runTestBuildPlanAutogenCreatesChartForUnmatchedGauge(t *testing.T) {
1364 + e, err := New(WithEnginePolicy(EnginePolicy{Autogen: &AutogenPolicy{Enabled: true}}))
1365 require.NoError(t, err)
1366
1367 yaml := `
@@ -1366,8 +1408,8 @@ groups:
1408 assert.Equal(t, float64(7), update.Values[0].Float64)
1409 }
1410
1369 -func TestBuildPlanAutogenCreatesChartForUnmatchedStateSet(t *testing.T) {
1370 - e, err := New(WithAutogenPolicy(AutogenPolicy{Enabled: true}))
1411 +func runTestBuildPlanAutogenCreatesChartForUnmatchedStateSet(t *testing.T) {
1412 + e, err := New(WithEnginePolicy(EnginePolicy{Autogen: &AutogenPolicy{Enabled: true}}))
1413 require.NoError(t, err)
1414
1415 yaml := `
@@ -1427,8 +1469,8 @@ groups:
1469 assert.Contains(t, dims, "operational")
1470 }
1471
1430 -func TestBuildPlanAutogenKeepsStateSetUnitsWhenMetricMetaUnitIsSet(t *testing.T) {
1431 - e, err := New(WithAutogenPolicy(AutogenPolicy{Enabled: true}))
1472 +func runTestBuildPlanAutogenKeepsStateSetUnitsWhenMetricMetaUnitIsSet(t *testing.T) {
1473 + e, err := New(WithEnginePolicy(EnginePolicy{Autogen: &AutogenPolicy{Enabled: true}}))
1474 require.NoError(t, err)
1475
1476 yaml := `
@@ -1472,8 +1514,8 @@ groups:
1514 assert.Equal(t, "state", create.Meta.Units)
1515 }
1516
1475 -func TestBuildPlanTemplateWinsOnAutogenChartIDCollisionAcrossSeries(t *testing.T) {
1476 - e, err := New(WithAutogenPolicy(AutogenPolicy{Enabled: true}))
1517 +func runTestBuildPlanTemplateWinsOnAutogenChartIDCollisionAcrossSeries(t *testing.T) {
1518 + e, err := New(WithEnginePolicy(EnginePolicy{Autogen: &AutogenPolicy{Enabled: true}}))
1519 require.NoError(t, err)
1520
1521 yaml := `
@@ -1521,11 +1563,11 @@ groups:
1563 assert.Equal(t, float64(7), update.Values[0].Float64)
1564 }
1565
1524 -func TestBuildPlanAutogenRemovalLifecycleExpiry(t *testing.T) {
1525 - e, err := New(WithAutogenPolicy(AutogenPolicy{
1566 +func runTestBuildPlanAutogenRemovalLifecycleExpiry(t *testing.T) {
1567 + e, err := New(WithEnginePolicy(EnginePolicy{Autogen: &AutogenPolicy{
1568 Enabled: true,
1569 ExpireAfterSuccessCycles: 1,
1528 - }))
1570 + }}))
1571 require.NoError(t, err)
1572
1573 yaml := `
@@ -1564,7 +1606,7 @@ groups:
1606 assert.Equal(t, []ActionKind{ActionRemoveChart}, actionKinds(plan2.Actions))
1607 }
1608
1567 -func TestBuildPlanFirstWriterWinsAndAccumulatesRepeatedRoutes(t *testing.T) {
1609 +func runTestBuildPlanFirstWriterWinsAndAccumulatesRepeatedRoutes(t *testing.T) {
1610 e, err := New()
1611 require.NoError(t, err)
1612
@@ -1637,7 +1679,7 @@ groups:
1679 assert.Equal(t, float64(8), update.Values[0].Float64)
1680 }
1681
1640 -func TestBuildPlanEmptyEmissionAndScratchReusePruneAcrossCycles(t *testing.T) {
1682 +func runTestBuildPlanEmptyEmissionAndScratchReusePruneAcrossCycles(t *testing.T) {
1683 e, err := New()
1684 require.NoError(t, err)
1685
@@ -1727,6 +1769,110 @@ groups:
1769 require.Contains(t, matChart.scratchEntries, "ok")
1770 }
1771
1772 +func TestBuildPlanSequenceModeScenarios(t *testing.T) {
1773 + tests := map[string]struct {
1774 + run func(t *testing.T)
1775 + }{
1776 + "collector mode keeps static success-seq dedupe semantics": {
1777 + run: func(t *testing.T) {
1778 + e, err := New()
1779 + require.NoError(t, err)
1780 + require.NoError(t, e.LoadYAML([]byte(`
1781 +version: v1
1782 +groups:
1783 + - family: Service
1784 + metrics:
1785 + - component.load
1786 + charts:
1787 + - id: component_load
1788 + title: Component Load
1789 + context: component_load
1790 + units: load
1791 + dimensions:
1792 + - selector: component.load
1793 + name: value
1794 +`), 1))
1795 +
1796 + store := metrix.NewCollectorStore()
1797 + cc := mustCycleController(t, store)
1798 + g := store.Write().SnapshotMeter("component").Gauge("load")
1799 +
1800 + cc.BeginCycle()
1801 + g.Observe(5)
1802 + cc.CommitCycleSuccess()
1803 +
1804 + plan1, err := e.BuildPlan(store.Read())
1805 + require.NoError(t, err)
1806 + require.NotNil(t, findUpdateAction(plan1))
1807 +
1808 + plan2, err := e.BuildPlan(store.Read())
1809 + require.NoError(t, err)
1810 + assert.Empty(t, plan2.Actions)
1811 + },
1812 + },
1813 + "runtime mode re-emits updates on no-write ticks and keeps scratch entries": {
1814 + run: func(t *testing.T) {
1815 + e, err := New(WithSeriesSelectionAllVisible(), WithRuntimePlannerMode())
1816 + require.NoError(t, err)
1817 + require.NoError(t, e.LoadYAML([]byte(`
1818 +version: v1
1819 +groups:
1820 + - family: Runtime
1821 + metrics:
1822 + - component.load
1823 + charts:
1824 + - id: component_load
1825 + title: Component Load
1826 + context: component_load
1827 + units: load
1828 + dimensions:
1829 + - selector: component.load
1830 + name_from_label: id
1831 +`), 1))
1832 +
1833 + store := metrix.NewRuntimeStore()
1834 + vec := store.Write().StatefulMeter("component").Vec("id").Gauge("load")
1835 + vec.WithLabelValues("ok").Set(1)
1836 + vec.WithLabelValues("warn").Set(2)
1837 +
1838 + reader := store.Read(metrix.ReadRaw(), metrix.ReadFlatten())
1839 + plan1, err := e.BuildPlan(reader)
1840 + require.NoError(t, err)
1841 + require.NotNil(t, findUpdateAction(plan1))
1842 +
1843 + matChart := e.state.materialized.charts["component_load"]
1844 + require.NotNil(t, matChart)
1845 + require.Contains(t, matChart.scratchEntries, "ok")
1846 + require.Contains(t, matChart.scratchEntries, "warn")
1847 + okEntry := matChart.scratchEntries["ok"]
1848 + require.NotNil(t, okEntry)
1849 +
1850 + plan2, err := e.BuildPlan(reader)
1851 + require.NoError(t, err)
1852 + assert.Equal(t, []ActionKind{ActionUpdateChart}, actionKinds(plan2.Actions))
1853 + require.NotNil(t, findUpdateAction(plan2))
1854 + metricsReader := e.RuntimeStore().Read(metrix.ReadRaw())
1855 + cacheHits, ok := metricsReader.Value("netdata.go.plugin.framework.chartengine.route_cache_hits_total", nil)
1856 + require.True(t, ok)
1857 + assert.GreaterOrEqual(t, cacheHits, float64(1))
1858 + fullDrops, fullDropsSeen := metricsReader.Value("netdata.go.plugin.framework.chartengine.route_cache_full_drops_total", nil)
1859 + require.True(t, fullDropsSeen)
1860 + assert.Equal(t, float64(0), fullDrops)
1861 +
1862 + matChart = e.state.materialized.charts["component_load"]
1863 + require.NotNil(t, matChart)
1864 + require.Contains(t, matChart.scratchEntries, "ok")
1865 + require.Contains(t, matChart.scratchEntries, "warn")
1866 + assert.Equal(t, okEntry, matChart.scratchEntries["ok"])
1867 + },
1868 + },
1869 + }
1870 +
1871 + for name, tc := range tests {
1872 + t.Run(name, tc.run)
1873 + }
1874 +}
1875 +
1876 func TestPlannerStageBoundaries(t *testing.T) {
1877 tests := map[string]func(t *testing.T){
1878 "scan stage accumulates per-chart state": func(t *testing.T) {
@@ -1767,7 +1913,8 @@ groups:
1913 InferredDimensions: make([]InferredDimension, 0),
1914 }
1915 reader := store.Read()
1770 - ctx, err := e.preparePlanBuildContext(reader, &out, reader.CollectMeta())
1916 + meta := reader.CollectMeta()
1917 + ctx, err := e.preparePlanBuildContext(reader, &out, meta, meta.LastSuccessSeq)
1918 require.NoError(t, err)
1919 require.NoError(t, e.scanPlanSeries(ctx))
1920
@@ -1816,7 +1963,8 @@ groups:
1963 InferredDimensions: make([]InferredDimension, 0),
1964 }
1965 reader := store.Read()
1819 - ctx, err := e.preparePlanBuildContext(reader, &out, reader.CollectMeta())
1966 + meta := reader.CollectMeta()
1967 + ctx, err := e.preparePlanBuildContext(reader, &out, meta, meta.LastSuccessSeq)
1968 require.NoError(t, err)
1969 require.NoError(t, e.scanPlanSeries(ctx))
1970 require.NoError(t, e.materializePlanCharts(ctx))
src/go/plugin/framework/chartengine/policy.go
+1 -3
@@ -4,7 +4,6 @@ package chartengine
4
5 import (
6 "fmt"
7 - "strings"
7
8 metrixselector "github.com/netdata/netdata/go/plugins/pkg/metrix/selector"
9 "github.com/netdata/netdata/go/plugins/plugin/framework/charttpl"
@@ -18,7 +17,6 @@ func resolveEffectivePolicy(cfg engineConfig, templatePolicy *charttpl.Engine) (
17 if templatePolicy.Autogen != nil {
18 normalized, err := normalizeAutogenPolicy(AutogenPolicy{
19 Enabled: templatePolicy.Autogen.Enabled,
21 - TypeID: strings.TrimSpace(templatePolicy.Autogen.TypeID),
20 MaxTypeIDLen: templatePolicy.Autogen.MaxTypeIDLen,
21 ExpireAfterSuccessCycles: templatePolicy.Autogen.ExpireAfterSuccessCycles,
22 })
@@ -29,7 +27,7 @@ func resolveEffectivePolicy(cfg engineConfig, templatePolicy *charttpl.Engine) (
27 }
28
29 if templatePolicy.Selector != nil {
32 - compiled, err := compileEngineSelector(templatePolicy.Selector)
30 + compiled, err := compileEngineSelector(*templatePolicy.Selector)
31 if err != nil {
32 return AutogenPolicy{}, nil, fmt.Errorf("template engine.selector: %w", err)
33 }
src/go/plugin/framework/chartengine/runtime_metrics.go
+16 -16
@@ -135,19 +135,19 @@ func newRuntimeMetrics(store metrix.RuntimeStore) *runtimeMetrics {
135 metrix.WithUnit("transitions"),
136 )
137 return &runtimeMetrics{
138 - buildSuccessTotal: meter.Counter(
138 + buildSuccessTotal: metrix.SeededCounter(meter,
139 "build_success_total",
140 metrix.WithDescription("Successful BuildPlan calls"),
141 metrix.WithChartFamily("ChartEngine/Build"),
142 metrix.WithUnit("builds"),
143 ),
144 - buildErrorTotal: meter.Counter(
144 + buildErrorTotal: metrix.SeededCounter(meter,
145 "build_error_total",
146 metrix.WithDescription("Failed BuildPlan calls"),
147 metrix.WithChartFamily("ChartEngine/Build"),
148 metrix.WithUnit("builds"),
149 ),
150 - buildSkippedFailedTotal: meter.Counter(
150 + buildSkippedFailedTotal: metrix.SeededCounter(meter,
151 "build_skipped_failed_collect_total",
152 metrix.WithDescription("BuildPlan calls skipped due failed collect cycle"),
153 metrix.WithChartFamily("ChartEngine/Build"),
@@ -171,69 +171,69 @@ func newRuntimeMetrics(store metrix.RuntimeStore) *runtimeMetrics {
171
172 buildSeqBrokenTotal: buildSeqTransitions.WithLabelValues("broken"),
173 buildSeqRecoveredTotal: buildSeqTransitions.WithLabelValues("recovered"),
174 - buildSeqViolation: meter.Gauge(
174 + buildSeqViolation: metrix.SeededGauge(meter,
175 "build_seq_violation_active",
176 metrix.WithDescription("1 when build sequence monotonicity is currently violated"),
177 metrix.WithChartFamily("ChartEngine/Build"),
178 metrix.WithUnit("state"),
179 ),
180
181 - routeCacheHitsTotal: meter.Counter(
181 + routeCacheHitsTotal: metrix.SeededCounter(meter,
182 "route_cache_hits_total",
183 metrix.WithDescription("Route cache lookup hits"),
184 metrix.WithChartFamily("ChartEngine/Route Cache"),
185 metrix.WithUnit("hits"),
186 ),
187 - routeCacheMissesTotal: meter.Counter(
187 + routeCacheMissesTotal: metrix.SeededCounter(meter,
188 "route_cache_misses_total",
189 metrix.WithDescription("Route cache lookup misses"),
190 metrix.WithChartFamily("ChartEngine/Route Cache"),
191 metrix.WithUnit("misses"),
192 ),
193 - routeCacheEntries: meter.Gauge(
193 + routeCacheEntries: metrix.SeededGauge(meter,
194 "route_cache_entries",
195 metrix.WithDescription("Current number of route cache entries"),
196 metrix.WithChartFamily("ChartEngine/Route Cache"),
197 metrix.WithUnit("entries"),
198 ),
199 - routeCacheRetainedTotal: meter.Counter(
199 + routeCacheRetainedTotal: metrix.SeededCounter(meter,
200 "route_cache_retained_total",
201 metrix.WithDescription("Route cache entries retained after prune"),
202 metrix.WithChartFamily("ChartEngine/Route Cache"),
203 metrix.WithUnit("entries"),
204 ),
205 - routeCachePrunedTotal: meter.Counter(
205 + routeCachePrunedTotal: metrix.SeededCounter(meter,
206 "route_cache_pruned_total",
207 metrix.WithDescription("Route cache entries pruned"),
208 metrix.WithChartFamily("ChartEngine/Route Cache"),
209 metrix.WithUnit("entries"),
210 ),
211 - routeCacheFullDropsTotal: meter.Counter(
211 + routeCacheFullDropsTotal: metrix.SeededCounter(meter,
212 "route_cache_full_drops_total",
213 metrix.WithDescription("Route cache full-drop prune events"),
214 metrix.WithChartFamily("ChartEngine/Route Cache"),
215 metrix.WithUnit("events"),
216 ),
217
218 - seriesScannedTotal: meter.Counter(
218 + seriesScannedTotal: metrix.SeededCounter(meter,
219 "series_scanned_total",
220 metrix.WithDescription("Series scanned by planner"),
221 metrix.WithChartFamily("ChartEngine/Series"),
222 metrix.WithUnit("series"),
223 ),
224 - seriesMatchedTotal: meter.Counter(
224 + seriesMatchedTotal: metrix.SeededCounter(meter,
225 "series_matched_total",
226 metrix.WithDescription("Series matched by template or autogen"),
227 metrix.WithChartFamily("ChartEngine/Series"),
228 metrix.WithUnit("series"),
229 ),
230 - seriesUnmatchedTotal: meter.Counter(
230 + seriesUnmatchedTotal: metrix.SeededCounter(meter,
231 "series_unmatched_total",
232 metrix.WithDescription("Series left unmatched after routing"),
233 metrix.WithChartFamily("ChartEngine/Series"),
234 metrix.WithUnit("series"),
235 ),
236 - seriesAutogenMatchedTotal: meter.Counter(
236 + seriesAutogenMatchedTotal: metrix.SeededCounter(meter,
237 "series_autogen_matched_total",
238 metrix.WithDescription("Series matched by autogen fallback"),
239 metrix.WithChartFamily("ChartEngine/Series"),
@@ -242,13 +242,13 @@ func newRuntimeMetrics(store metrix.RuntimeStore) *runtimeMetrics {
242 seriesFilteredBySeq: seriesFiltered.WithLabelValues("by_seq"),
243 seriesFilteredBySelector: seriesFiltered.WithLabelValues("by_selector"),
244
245 - planChartInstances: meter.Gauge(
245 + planChartInstances: metrix.SeededGauge(meter,
246 "plan_chart_instances",
247 metrix.WithDescription("Chart instances in last successful build plan"),
248 metrix.WithChartFamily("ChartEngine/Plan"),
249 metrix.WithUnit("charts"),
250 ),
251 - planInferredDimensions: meter.Gauge(
251 + planInferredDimensions: metrix.SeededGauge(meter,
252 "plan_inferred_dimensions",
253 metrix.WithDescription("Inferred dimensions in last successful build plan"),
254 metrix.WithChartFamily("ChartEngine/Plan"),
src/go/plugin/framework/chartengine/runtime_metrics_test.go
+7 -2
@@ -178,7 +178,11 @@ func TestEngineRuntimeObservabilityScenarios(t *testing.T) {
178 component := rs.Write().StatefulMeter("component").Counter("jobs_total")
179 component.Add(7)
180
181 - observer, err := New(WithRuntimeStore(nil))
181 + observer, err := New(
182 + WithRuntimeStore(nil),
183 + WithSeriesSelectionAllVisible(),
184 + WithRuntimePlannerMode(),
185 + )
186 require.NoError(t, err)
187 require.NoError(t, observer.LoadYAML([]byte(runtimeComponentTemplateYAML()), 1))
188
@@ -212,8 +216,9 @@ func TestEngineRuntimeObservabilityScenarios(t *testing.T) {
216
217 observer, err := New(
218 WithRuntimeStore(nil),
215 - WithAutogenPolicy(AutogenPolicy{Enabled: true}),
219 + WithEnginePolicy(EnginePolicy{Autogen: &AutogenPolicy{Enabled: true}}),
220 WithSeriesSelectionAllVisible(),
221 + WithRuntimePlannerMode(),
222 )
223 require.NoError(t, err)
224 require.NoError(t, observer.LoadYAML([]byte(runtimeDummyTemplateYAML()), 1))
src/go/plugin/framework/chartengine/state.go
+7 -4
@@ -16,10 +16,13 @@ type engineState struct {
16 materialized materializedState
17 hints plannerSizingHints
18 buildSeq buildSeqState
19 - stats engineStats
20 - runtimeStore metrix.RuntimeStore
21 - runtimeStats *runtimeMetrics
22 - log *logger.Logger
19 + // plannerBuildSeq is runtime-mode build-cycle sequence used only by
20 + // per-build dedupe/scratch bookkeeping.
21 + plannerBuildSeq uint64
22 + stats engineStats
23 + runtimeStore metrix.RuntimeStore
24 + runtimeStats *runtimeMetrics
25 + log *logger.Logger
26 }
27
28 type plannerSizingHints struct {
src/go/plugin/framework/charttpl/README.md
-1
@@ -106,7 +106,6 @@ The metric name prefix is required; label-only selectors like `{label=value}` ar
106 | `selector.allow` | array[string] | Global include selectors |
107 | `selector.deny` | array[string] | Global exclude selectors |
108 | `autogen.enabled` | bool | Enable unmatched-series autogen fallback |
109 -| `autogen.type_id` | string | `type.id` prefix budget base |
109 | `autogen.max_type_id_len` | int | Max full `type.id` length (`0` = default; must be `0` or `>= 4` when set) |
110 | `autogen.expire_after_success_cycles` | uint64 | Autogen lifecycle expiry |
111
src/go/plugin/framework/charttpl/config_schema.json
-3
@@ -54,9 +54,6 @@
54 "enabled": {
55 "type": "boolean"
56 },
57 - "type_id": {
58 - "type": "string"
59 - },
57 "max_type_id_len": {
58 "oneOf": [
59 {
src/go/plugin/framework/charttpl/spec.go
-3
@@ -25,9 +25,6 @@ type Engine struct {
25 type EngineAutogen struct {
26 Enabled bool `yaml:"enabled,omitempty" json:"enabled,omitempty"`
27
28 - // TypeID is the chart-type prefix used by Netdata runtime checks
29 - // (`type.id` length guard). Typically this is `<plugin>.<job>`.
30 - TypeID string `yaml:"type_id,omitempty" json:"type_id,omitempty"`
28 // MaxTypeIDLen is the max allowed full `type.id` length.
29 // Zero means default (1200).
30 MaxTypeIDLen int `yaml:"max_type_id_len,omitempty" json:"max_type_id_len,omitempty"`
src/go/plugin/framework/charttpl/spec_test.go
-2
@@ -28,7 +28,6 @@ engine:
28 - mysql_queries_total{db="main"}
29 autogen:
30 enabled: true
31 - type_id: mysql.jobs
31 max_type_id_len: 512
32 expire_after_success_cycles: 9
33 groups:
@@ -65,7 +64,6 @@ groups:
64 assert.Equal(t, []string{`mysql_queries_total{db="main"}`}, spec.Engine.Selector.Allow)
65 require.NotNil(t, spec.Engine.Autogen)
66 assert.True(t, spec.Engine.Autogen.Enabled)
68 - assert.Equal(t, "mysql.jobs", spec.Engine.Autogen.TypeID)
67 assert.Equal(t, 512, spec.Engine.Autogen.MaxTypeIDLen)
68 assert.Equal(t, uint64(9), spec.Engine.Autogen.ExpireAfterSuccessCycles)
69 },
src/go/plugin/framework/functions/manager.go
+1
@@ -176,6 +176,7 @@ func (m *Manager) run(ctx context.Context, quitCh chan struct{}) {
176 m.handleCancelEvent(event)
177 continue
178 case inputEventCall:
179 + m.observeFunctionCall()
180 m.dispatchInvocation(ctx, event.fn)
181 }
182 }
src/go/plugin/framework/functions/manager_flow_test.go
+24 -16
@@ -260,29 +260,27 @@ func TestManager_FlowScenarios(t *testing.T) {
260 run: func(t *testing.T, mgr *Manager, in *chanInput, out *safeBuffer) {
261 mgr.workerCount = 2
262 started := make(chan struct{}, 1)
263 + tx2Started := make(chan struct{}, 1)
264 release := make(chan struct{})
264 - var (
265 - calls atomic.Int32
266 - running atomic.Int32
267 - maxSeen atomic.Int32
268 - )
265 + var calls atomic.Int32
266
267 mgr.Register("fn", func(fn Function) {
268 calls.Add(1)
269
273 - curr := running.Add(1)
274 - for {
275 - prev := maxSeen.Load()
276 - if curr <= prev || maxSeen.CompareAndSwap(prev, curr) {
277 - break
278 - }
279 - }
280 - defer running.Add(-1)
281 -
270 if fn.UID == "tx1" {
271 started <- struct{}{}
272 <-release
273 + mgr.respUID(fn.UID, 200, "ok")
274 + return
275 }
276 +
277 + if fn.UID == "tx2" {
278 + select {
279 + case tx2Started <- struct{}{}:
280 + default:
281 + }
282 + }
283 +
284 mgr.respUID(fn.UID, 200, "ok")
285 })
286
@@ -295,7 +293,19 @@ func TestManager_FlowScenarios(t *testing.T) {
293 in.ch <- functionLine("tx2", "fn")
294 // Duplicate active UID must not advance lanes or finalize tx1.
295 in.ch <- functionLine("tx1", "fn")
296 +
297 + select {
298 + case <-tx2Started:
299 + t.Fatal("tx2 started before tx1 was released")
300 + default:
301 + }
302 +
303 close(release)
304 + select {
305 + case <-tx2Started:
306 + case <-time.After(time.Second):
307 + t.Fatal("tx2 did not start after tx1 was released")
308 + }
309 close(in.ch)
310 waitForDone(t, done)
311
@@ -304,8 +314,6 @@ func TestManager_FlowScenarios(t *testing.T) {
314 assert.Equal(t, 1, strings.Count(got, "FUNCTION_RESULT_BEGIN tx2 200"))
315 assert.Equal(t, 0, strings.Count(got, "FUNCTION_RESULT_BEGIN tx1 409"))
316 assert.EqualValues(t, 2, calls.Load())
307 - // Same key must remain serialized despite duplicate UID input.
308 - assert.EqualValues(t, 1, maxSeen.Load())
317 },
318 },
319 "duplicate tombstoned uid is ignored without extra terminal output": {
src/go/plugin/framework/functions/runtime_metrics.go
+21 -11
@@ -11,6 +11,7 @@ type managerRuntimeMetrics struct {
11 invocationsAwaitingResult metrix.StatefulGauge
12 schedulerPending metrix.StatefulGauge
13
14 + functionCallsTotal metrix.StatefulCounter
15 queueFullTotal metrix.StatefulCounter
16 cancelFallbackTotal metrix.StatefulCounter
17 lateTerminalDropped metrix.StatefulCounter
@@ -24,43 +25,49 @@ func newManagerRuntimeMetrics(store metrix.RuntimeStore) *managerRuntimeMetrics
25
26 meter := store.Write().StatefulMeter(functionsRuntimeMetricPrefix)
27 metrics := &managerRuntimeMetrics{
27 - invocationsActive: meter.Gauge(
28 + invocationsActive: metrix.SeededGauge(meter,
29 "invocations_active",
30 metrix.WithDescription("Current number of active function invocations tracked by UID"),
31 metrix.WithChartFamily("Framework/Functions/Invocations"),
32 metrix.WithUnit("invocations"),
33 ),
33 - invocationsAwaitingResult: meter.Gauge(
34 + invocationsAwaitingResult: metrix.SeededGauge(meter,
35 "invocations_awaiting_result",
36 metrix.WithDescription("Current number of active invocations waiting for terminal response"),
37 metrix.WithChartFamily("Framework/Functions/Invocations"),
38 metrix.WithUnit("invocations"),
39 ),
39 - schedulerPending: meter.Gauge(
40 + schedulerPending: metrix.SeededGauge(meter,
41 "scheduler_pending",
42 metrix.WithDescription("Current number of invocations pending in scheduler"),
43 metrix.WithChartFamily("Framework/Functions/Scheduler"),
44 metrix.WithUnit("invocations"),
45 ),
45 - queueFullTotal: meter.Counter(
46 + functionCallsTotal: metrix.SeededCounter(meter,
47 + "calls_total",
48 + metrix.WithDescription("Total number of parsed function call requests"),
49 + metrix.WithChartFamily("Framework/Functions/Calls"),
50 + metrix.WithUnit("calls"),
51 + ),
52 + queueFullTotal: metrix.SeededCounter(meter,
53 "queue_full_total",
54 metrix.WithDescription("Total number of function requests rejected due to queue full"),
55 metrix.WithChartFamily("Framework/Functions/Failures"),
56 metrix.WithUnit("requests"),
57 ),
51 - cancelFallbackTotal: meter.Counter(
58 + cancelFallbackTotal: metrix.SeededCounter(meter,
59 "cancel_fallback_total",
60 metrix.WithDescription("Total number of function requests finalized by cancel fallback timer"),
61 metrix.WithChartFamily("Framework/Functions/Cancellation"),
62 metrix.WithUnit("requests"),
63 ),
57 - lateTerminalDropped: meter.Counter(
64 + lateTerminalDropped: metrix.SeededCounter(meter,
65 "late_terminal_dropped_total",
66 metrix.WithDescription("Total number of late terminal responses dropped by tombstone guard"),
67 metrix.WithChartFamily("Framework/Functions/Finalization"),
68 metrix.WithUnit("responses"),
69 ),
63 - duplicateUIDIgnored: meter.Counter(
70 + duplicateUIDIgnored: metrix.SeededCounter(meter,
71 "duplicate_uid_ignored_total",
72 metrix.WithDescription("Total number of duplicate transaction IDs ignored at admission"),
73 metrix.WithChartFamily("Framework/Functions/Admission"),
@@ -68,10 +75,6 @@ func newManagerRuntimeMetrics(store metrix.RuntimeStore) *managerRuntimeMetrics
75 ),
76 }
77
71 - metrics.invocationsActive.Set(0)
72 - metrics.invocationsAwaitingResult.Set(0)
73 - metrics.schedulerPending.Set(0)
74 -
78 return metrics
79 }
80
@@ -106,6 +109,13 @@ func (m *Manager) observeQueueFull() {
109 m.runtimeMetrics.queueFullTotal.Add(1)
110 }
111
112 +func (m *Manager) observeFunctionCall() {
113 + if m == nil || m.runtimeMetrics == nil {
114 + return
115 + }
116 + m.runtimeMetrics.functionCallsTotal.Add(1)
117 +}
118 +
119 func (m *Manager) observeCancelFallback() {
120 if m == nil || m.runtimeMetrics == nil {
121 return
src/go/plugin/framework/functions/runtime_metrics_test.go
+1
@@ -125,6 +125,7 @@ func TestManager_RuntimeMetricsScenarios(t *testing.T) {
125 assert.GreaterOrEqual(t, runtimeMetricValue(t, mgr.runtimeStore, functionsRuntimeMetricPrefix+".cancel_fallback_total", nil), float64(1))
126 assert.GreaterOrEqual(t, runtimeMetricValue(t, mgr.runtimeStore, functionsRuntimeMetricPrefix+".late_terminal_dropped_total", nil), float64(1))
127 assert.GreaterOrEqual(t, runtimeMetricValue(t, mgr.runtimeStore, functionsRuntimeMetricPrefix+".duplicate_uid_ignored_total", nil), float64(2))
128 + assert.Equal(t, float64(5), runtimeMetricValue(t, mgr.runtimeStore, functionsRuntimeMetricPrefix+".calls_total", nil))
129
130 assert.Equal(t, float64(0), runtimeMetricValue(t, mgr.runtimeStore, functionsRuntimeMetricPrefix+".invocations_active", nil))
131 assert.Equal(t, float64(0), runtimeMetricValue(t, mgr.runtimeStore, functionsRuntimeMetricPrefix+".invocations_awaiting_result", nil))
src/go/plugin/framework/jobruntime/job_v2.go
+2 -5
@@ -302,13 +302,10 @@ func (j *JobV2) postCheck() error {
302
303 opts := []chartengine.Option{
304 chartengine.WithLogger(j.Logger.With(slog.String("component", "chartengine"))),
305 + chartengine.WithEmitTypeIDBudgetPrefix(j.fullName),
306 }
307 if v, ok := j.module.(collectorapi.CollectorV2EnginePolicy); ok {
307 - policy := v.EnginePolicy()
308 - // Chartengine autogen type.id budget must use the actual emitted type.id.
309 - // JobV2 always emits with fullName as TypeID.
310 - policy.Autogen.TypeID = j.fullName
311 - opts = append(opts, chartengine.WithEnginePolicy(policy))
308 + opts = append(opts, chartengine.WithEnginePolicy(v.EnginePolicy()))
309 }
310
311 engine, err := chartengine.New(opts...)
src/go/plugin/framework/runtimecomp/types.go
-3
@@ -9,9 +9,6 @@ import "github.com/netdata/netdata/go/plugins/pkg/metrix"
9 type AutogenPolicy struct {
10 Enabled bool
11
12 - // TypeID is the chart-type prefix used by Netdata runtime checks
13 - // (`type.id` length guard). Typically this is `<plugin>.<job>`.
14 - TypeID string
12 // MaxTypeIDLen is the max allowed full `type.id` length.
13 // Zero means default (1200).
14 MaxTypeIDLen int
src/go/plugin/go.d/pkg/ndexec/ndexec_test.go
+1 -1
@@ -65,7 +65,7 @@ exec "$@"
65 "success_echo_args": {
66 helperPath: helper,
67 argv: []string{echoArgs, `a b`, `c"d`},
68 - timeout: time.Second,
68 + timeout: 15 * time.Second,
69 wantOut: "a b|c\"d|\n",
70 },
71 "nonzero_with_trimmed_stderr": {