improve(go.d/ddsnmp): add group_by for virtual metrics (#20970)
Ilya Mashchenko committed
Sep 13, 2025 at 19:13 UTC
4830ddff1c26b8449848e766e5677b9bf4eea4d8
3 files changed
+563
-108
src/go/plugin/go.d/collector/snmp/ddsnmp/ddprofiledefinition/virtual_metrics.go
+2
@@ -6,6 +6,7 @@ import (
6
7
type VirtualMetricConfig struct {
8
Name string `yaml:"name"`
9
+ GroupBy []string `yaml:"group_by"`
10
Sources []VirtualMetricSourceConfig `yaml:"sources"`
11
ChartMeta ChartMeta `yaml:"chart_meta"`
12
}
@@ -13,6 +14,7 @@ type VirtualMetricConfig struct {
14
func (vm VirtualMetricConfig) Clone() VirtualMetricConfig {
15
return VirtualMetricConfig{
16
Name: vm.Name,
17
+ GroupBy: slices.Clone(vm.GroupBy),
18
Sources: slices.Clone(vm.Sources),
19
ChartMeta: vm.ChartMeta,
20
}
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/collector_vmetrics.go
+319
-108
@@ -4,6 +4,8 @@ package ddsnmpcollector
4
5
import (
6
"slices"
7
+ "sort"
8
+ "strings"
9
10
"github.com/netdata/netdata/go/plugins/logger"
11
"github.com/netdata/netdata/go/plugins/plugin/go.d/collector/snmp/ddsnmp"
@@ -20,102 +22,168 @@ func newVirtualMetricsCollector(log *logger.Logger) *vmetricsCollector {
22
}
23
}
24
23
-type (
24
- // vmetricsSourceKey identifies a metric source
25
- vmetricsSourceKey struct {
26
- metricName string
27
- tableName string
28
- }
29
- // which aggregator to feed and under which dimension name
30
- vmetricsSink struct {
31
- agg *vmetricsAggregator
32
- dim string // empty == non-composite (single-source behavior)
33
- }
34
- // vmetricsAggregator holds accumulation state for a virtual metric
35
- vmetricsAggregator struct {
36
- config ddprofiledefinition.VirtualMetricConfig
37
- sum int64
38
- multiSum map[string]int64 // for aggregating MultiValue metrics
39
- perDim map[string]int64 // per-source accumulation when composite
40
- sourceCount int
41
- metricType ddprofiledefinition.ProfileMetricType
42
- }
43
-)
44
-
45
-func (p *vmetricsCollector) Collect(profDef *ddprofiledefinition.ProfileDefinition, collectedMetrics []ddsnmp.Metric) []ddsnmp.Metric {
25
+func (p *vmetricsCollector) Collect(profDef *ddprofiledefinition.ProfileDefinition, collected []ddsnmp.Metric) []ddsnmp.Metric {
26
if len(profDef.VirtualMetrics) == 0 {
27
return nil
28
}
29
+ lookup, aggrs := p.buildAggregators(profDef)
30
+ p.accumulate(lookup, collected)
31
+ return p.emit(aggrs)
32
+}
33
50
- sourceToAggregators, aggregators := p.buildAggregators(profDef)
51
-
52
- for _, metric := range collectedMetrics {
53
- key := vmetricsSourceKey{
54
- metricName: metric.Name,
55
- tableName: metric.Table,
56
- }
57
-
58
- sinks, found := sourceToAggregators[key]
59
- if !found {
34
+func (p *vmetricsCollector) accumulate(lookup map[vmetricsSourceKey][]vmetricsSink, collected []ddsnmp.Metric) {
35
+ for _, m := range collected {
36
+ sinks, ok := lookup[vmetricsSourceKey{metricName: m.Name, tableName: m.Table}]
37
+ if !ok {
38
continue
39
}
40
63
- for _, sink := range sinks {
64
- if sink.agg == nil {
65
- continue
66
- }
41
+ v, mv := vmCollapseMetricValue(m)
42
43
+ for _, sink := range sinks {
44
agg := sink.agg
69
-
70
- // If sink.dim != "" => composite path (this VM has multiple sources)
71
- if sink.dim != "" {
72
- // We need a single number per source (dimension).
73
- // If the incoming base metric is MultiValue, collapse it to a total; else use Value.
74
- var v int64
75
- if len(metric.MultiValue) > 0 {
76
- for _, mv := range metric.MultiValue {
77
- v += mv
78
- }
79
- } else {
80
- v = metric.Value
81
- }
82
-
83
- if agg.perDim == nil {
84
- agg.perDim = make(map[string]int64)
85
- }
86
- agg.perDim[sink.dim] += v
87
- } else {
88
- // Non-composite (single-source) => keep existing behavior:
89
- if len(metric.MultiValue) > 0 {
90
- if agg.multiSum == nil {
91
- agg.multiSum = make(map[string]int64)
92
- }
93
- for state, value := range metric.MultiValue {
94
- agg.multiSum[state] += value
95
- }
96
- } else {
97
- agg.sum += metric.Value
98
- }
45
+ if agg == nil {
46
+ continue
47
}
48
101
- agg.sourceCount++
49
if agg.metricType == "" {
103
- agg.metricType = metric.MetricType
104
- } else if agg.metricType != metric.MetricType {
105
- p.log.Debugf("virtual metric %q mixes MetricType (%s vs %s); using %s",
106
- agg.config.Name, agg.metricType, metric.MetricType, agg.metricType)
50
+ agg.metricType = m.MetricType
51
}
52
+
53
+ agg.accumulate(sink, v, mv, m.Tags)
54
}
55
}
56
+}
57
111
- // Build virtual metrics from aggregators
112
- var virtualMetrics []ddsnmp.Metric
113
- for _, agg := range aggregators {
58
+func (p *vmetricsCollector) emit(aggrs []*vmetricsAggregator) []ddsnmp.Metric {
59
+ // small pre-alloc: 1 total or ~groups count; start conservative
60
+ out := make([]ddsnmp.Metric, 0, len(aggrs))
61
+ for _, agg := range aggrs {
62
if agg.sourceCount == 0 {
63
p.log.Debugf("no source metrics found for virtual metric '%s'", agg.config.Name)
64
continue
65
}
66
+ agg.emitInto(&out)
67
+ }
68
+ return out
69
+}
70
+
71
+type (
72
+ // vmetricsAggregator holds accumulation state for a virtual metric
73
+ vmetricsAggregator struct {
74
+ config ddprofiledefinition.VirtualMetricConfig
75
+ metricType ddprofiledefinition.ProfileMetricType
76
+
77
+ // --- grouping controls ---
78
+ grouped bool // len(GroupBy) > 0
79
+ perRow bool // GroupBy == ["*"]
80
+ groupBy []string // explicit labels (nil for perRow/none)
81
+ groupTable string // v1: sources must share same table
82
+ perGroup map[string]*vmetricsGroupBucket
83
+
84
+ // --- dimensions (composite) ---
85
+ dims vmetricsDimSpec
86
+
87
+ // --- non-grouped accumulators (existing behavior) ---
88
+ perDim map[string]int64 // composite total (dim -> sum)
89
+ sum int64 // single-source total
90
+ multiSum map[string]int64 // merged base MultiValue
91
+ sourceCount int
92
+
93
+ keyBuf strings.Builder
94
+ }
95
+
96
+ // vmetricsSourceKey identifies a metric source
97
+ vmetricsSourceKey struct {
98
+ metricName string
99
+ tableName string
100
+ }
101
+
102
+ // sink binding; dimIdx == -1 for non-composite
103
+ vmetricsSink struct {
104
+ agg *vmetricsAggregator
105
+ dimIdx int16
106
+ }
107
+
108
+ // precomputed dimension spec (avoids per-sample string lookups)
109
+ vmetricsDimSpec struct {
110
+ names []string
111
+ idxByName map[string]int // build-time only
112
+ count int
113
+ }
114
115
+ // per-group accumulator (emitted as one table row)
116
+ vmetricsGroupBucket struct {
117
+ vals []int64 // len == dims.count when composite
118
+ seen []bool
119
+ sum int64 // single-source grouped case
120
+ emitTags map[string]string // explicit group_by: tiny map; per-row "*": pointer to source Tags
121
+ }
122
+)
123
+
124
+func (agg *vmetricsAggregator) accumulate(sink vmetricsSink, v int64, mv map[string]int64, tags map[string]string) {
125
+ if agg.grouped {
126
+ agg.accumulateGrouped(sink, v, tags)
127
+ } else {
128
+ agg.accumulateTotal(sink, v, mv)
129
+ }
130
+ agg.sourceCount++
131
+}
132
+
133
+func (agg *vmetricsAggregator) accumulateGrouped(sink vmetricsSink, v int64, tags map[string]string) {
134
+ gkey, ok := vmBuildGroupKey(tags, agg)
135
+ if !ok {
136
+ return
137
+ }
138
+ b := agg.perGroup[gkey]
139
+ if b == nil {
140
+ b = &vmetricsGroupBucket{emitTags: vmBuildEmitTags(tags, agg)}
141
+ if agg.dims.count > 0 {
142
+ b.vals = make([]int64, agg.dims.count)
143
+ b.seen = make([]bool, agg.dims.count)
144
+ }
145
+ agg.perGroup[gkey] = b
146
+ }
147
+ if sink.dimIdx >= 0 {
148
+ b.vals[sink.dimIdx] += v
149
+ b.seen[sink.dimIdx] = true
150
+ } else {
151
+ b.sum += v
152
+ }
153
+}
154
+
155
+func (agg *vmetricsAggregator) accumulateTotal(sink vmetricsSink, v int64, mv map[string]int64) {
156
+ if sink.dimIdx >= 0 {
157
+ if agg.perDim == nil {
158
+ agg.perDim = make(map[string]int64, agg.dims.count)
159
+ }
160
+ name := agg.dims.names[sink.dimIdx]
161
+ agg.perDim[name] += v
162
+ return
163
+ }
164
+ // Single-source path: preserve MultiValue if provided
165
+ if mv != nil {
166
+ if agg.multiSum == nil {
167
+ agg.multiSum = make(map[string]int64, len(mv))
168
+ }
169
+ for k, x := range mv {
170
+ agg.multiSum[k] += x
171
+ }
172
+ return
173
+ }
174
+ agg.sum += v
175
+}
176
+
177
+func (agg *vmetricsAggregator) emitInto(out *[]ddsnmp.Metric) {
178
+ if agg.grouped {
179
+ agg.emitGrouped(out)
180
+ } else {
181
+ agg.emitTotal(out)
182
+ }
183
+}
184
+
185
+func (agg *vmetricsAggregator) emitGrouped(out *[]ddsnmp.Metric) {
186
+ for _, b := range agg.perGroup {
187
vm := ddsnmp.Metric{
188
Name: agg.config.Name,
189
Description: agg.config.ChartMeta.Description,
@@ -123,69 +191,131 @@ func (p *vmetricsCollector) Collect(profDef *ddprofiledefinition.ProfileDefiniti
191
Unit: agg.config.ChartMeta.Unit,
192
ChartType: agg.config.ChartMeta.Type,
193
MetricType: agg.metricType,
194
+ IsTable: true,
195
+ Table: agg.groupTable,
196
+ Tags: b.emitTags,
197
}
127
-
128
- switch {
129
- case len(agg.perDim) > 0:
130
- // Composite output: one metric with MultiValue where keys are dimension names (sources)
131
- vm.MultiValue = agg.perDim
132
- case len(agg.multiSum) > 0:
133
- // Single-source whose base metric was MultiValue
134
- vm.MultiValue = agg.multiSum
135
- default:
136
- // Single-source
137
- vm.Value = agg.sum
198
+ if agg.dims.count > 0 {
199
+ mv := make(map[string]int64, agg.dims.count)
200
+ for i, dn := range agg.dims.names {
201
+ if b.seen[i] {
202
+ mv[dn] = b.vals[i]
203
+ }
204
+ }
205
+ vm.MultiValue = mv
206
+ } else {
207
+ vm.Value = b.sum
208
}
139
- virtualMetrics = append(virtualMetrics, vm)
209
+ *out = append(*out, vm)
210
}
211
+}
212
142
- return virtualMetrics
213
+func (agg *vmetricsAggregator) emitTotal(out *[]ddsnmp.Metric) {
214
+ vm := ddsnmp.Metric{
215
+ Name: agg.config.Name,
216
+ Description: agg.config.ChartMeta.Description,
217
+ Family: agg.config.ChartMeta.Family,
218
+ Unit: agg.config.ChartMeta.Unit,
219
+ ChartType: agg.config.ChartMeta.Type,
220
+ MetricType: agg.metricType,
221
+ }
222
+ switch {
223
+ case len(agg.perDim) > 0:
224
+ vm.MultiValue = agg.perDim
225
+ case len(agg.multiSum) > 0:
226
+ vm.MultiValue = agg.multiSum
227
+ default:
228
+ vm.Value = agg.sum
229
+ }
230
+ *out = append(*out, vm)
231
}
232
233
func (p *vmetricsCollector) buildAggregators(profDef *ddprofiledefinition.ProfileDefinition) (map[vmetricsSourceKey][]vmetricsSink, []*vmetricsAggregator) {
146
- sourceToAggregators := make(map[vmetricsSourceKey][]vmetricsSink)
234
+ sourceToSinks := make(map[vmetricsSourceKey][]vmetricsSink)
235
aggregators := make([]*vmetricsAggregator, 0, len(profDef.VirtualMetrics))
236
237
existingNames := p.getDefinedMetricNames(profDef.Metrics)
238
151
- for _, config := range profDef.VirtualMetrics {
152
- if existingNames[config.Name] {
153
- p.log.Warningf("virtual metric '%s' conflicts with existing metric, skipping", config.Name)
239
+ for _, cfg := range profDef.VirtualMetrics {
240
+ if existingNames[cfg.Name] {
241
+ p.log.Warningf("virtual metric '%s' conflicts with existing metric, skipping", cfg.Name)
242
continue
243
}
244
157
- agg := &vmetricsAggregator{config: config}
158
- aggregators = append(aggregators, agg)
245
+ agg := &vmetricsAggregator{config: cfg}
246
+
247
+ // --- grouping detection / validation ---
248
+ if len(cfg.GroupBy) > 0 {
249
+ agg.grouped = true
250
+ agg.perRow = slices.Contains(cfg.GroupBy, "*")
251
+ if !agg.perRow {
252
+ agg.groupBy = cfg.GroupBy
253
+ }
254
+
255
+ // require all sources from the same table
256
+ var table string
257
+ same := true
258
+ for i, s := range cfg.Sources {
259
+ if i == 0 {
260
+ table = s.Table
261
+ } else if s.Table != table {
262
+ same = false
263
+ break
264
+ }
265
+ }
266
+ if !same || table == "" {
267
+ p.log.Warningf("virtual metric '%s' uses group_by but sources span tables or have no table; skipping (no joins yet)", cfg.Name)
268
+ continue
269
+ }
270
+ agg.groupTable = table
271
+ agg.perGroup = make(map[string]*vmetricsGroupBucket, 64)
272
+ }
273
160
- isComposite := len(config.Sources) > 1 &&
161
- slices.ContainsFunc(config.Sources, func(s ddprofiledefinition.VirtualMetricSourceConfig) bool {
274
+ // --- composite dims? (multiple sources and at least one 'as') ---
275
+ isComposite := len(cfg.Sources) > 1 &&
276
+ slices.ContainsFunc(cfg.Sources, func(s ddprofiledefinition.VirtualMetricSourceConfig) bool {
277
return s.As != ""
278
})
279
165
- // Register this aggregator for each source it needs
166
- for _, source := range config.Sources {
167
- key := vmetricsSourceKey{
168
- metricName: source.Metric,
169
- tableName: source.Table,
280
+ if isComposite {
281
+ agg.dims.idxByName = make(map[string]int, len(cfg.Sources))
282
+ agg.dims.names = make([]string, 0, len(cfg.Sources))
283
+ for _, s := range cfg.Sources {
284
+ name := ternary(s.As != "", s.As, s.Metric)
285
+ if _, dup := agg.dims.idxByName[name]; !dup {
286
+ agg.dims.idxByName[name] = len(agg.dims.names)
287
+ agg.dims.names = append(agg.dims.names, name)
288
+ }
289
}
290
+ agg.dims.count = len(agg.dims.names)
291
+ }
292
172
- var dim string
293
+ // register sinks
294
+ for _, src := range cfg.Sources {
295
+ key := vmetricsSourceKey{metricName: src.Metric, tableName: src.Table}
296
+
297
+ dimIdx := int16(-1)
298
if isComposite {
174
- dim = ternary(source.As != "", source.As, source.Metric)
299
+ name := ternary(src.As != "", src.As, src.Metric)
300
+ if idx, ok := agg.dims.idxByName[name]; ok {
301
+ dimIdx = int16(idx)
302
+ }
303
}
304
177
- sourceToAggregators[key] = append(sourceToAggregators[key], vmetricsSink{
178
- agg: agg,
179
- dim: dim,
305
+ sourceToSinks[key] = append(sourceToSinks[key], vmetricsSink{
306
+ agg: agg,
307
+ dimIdx: dimIdx,
308
})
309
}
310
+
311
+ aggregators = append(aggregators, agg)
312
}
313
184
- return sourceToAggregators, aggregators
314
+ return sourceToSinks, aggregators
315
}
316
317
func (p *vmetricsCollector) getDefinedMetricNames(profMetrics []ddprofiledefinition.MetricsConfig) map[string]bool {
188
- names := make(map[string]bool)
318
+ names := make(map[string]bool, len(profMetrics))
319
for _, m := range profMetrics {
320
switch {
321
case m.IsScalar():
@@ -198,3 +328,84 @@ func (p *vmetricsCollector) getDefinedMetricNames(profMetrics []ddprofiledefinit
328
}
329
return names
330
}
331
+
332
+// vmBuildGroupKey returns a stable group key.
333
+func vmBuildGroupKey(tags map[string]string, agg *vmetricsAggregator) (string, bool) {
334
+ if !agg.grouped {
335
+ return "", false
336
+ }
337
+
338
+ const (
339
+ groupKeySep = '\x1F' // ASCII Unit Separator: safe delimiter between label values/pairs
340
+ kvSep = '=' // used only in per-row fallback "k=v"
341
+ )
342
+
343
+ if agg.perRow {
344
+ if len(tags) == 0 {
345
+ return "", false
346
+ }
347
+ keys := make([]string, 0, len(tags))
348
+ for k := range tags {
349
+ keys = append(keys, k)
350
+ }
351
+ sort.Strings(keys)
352
+ agg.keyBuf.Reset()
353
+ for i, k := range keys {
354
+ if i > 0 {
355
+ agg.keyBuf.WriteByte(groupKeySep)
356
+ }
357
+ agg.keyBuf.WriteString(k)
358
+ agg.keyBuf.WriteByte(kvSep)
359
+ agg.keyBuf.WriteString(tags[k])
360
+ }
361
+ return agg.keyBuf.String(), true
362
+ }
363
+
364
+ switch len(agg.groupBy) {
365
+ case 0:
366
+ return "", false
367
+ case 1:
368
+ v := tags[agg.groupBy[0]]
369
+ return v, v != ""
370
+ default:
371
+ agg.keyBuf.Reset()
372
+ for i, l := range agg.groupBy {
373
+ v := tags[l]
374
+ if v == "" {
375
+ return "", false
376
+ }
377
+ if i > 0 {
378
+ agg.keyBuf.WriteByte(groupKeySep)
379
+ }
380
+ agg.keyBuf.WriteString(v)
381
+ }
382
+ return agg.keyBuf.String(), true
383
+ }
384
+}
385
+
386
+// vmBuildEmitTags captures labels to emit for a group (called once per new group)
387
+func vmBuildEmitTags(tags map[string]string, agg *vmetricsAggregator) map[string]string {
388
+ if agg.perRow {
389
+ // per-row: reuse pointer; we never mutate it here
390
+ return tags
391
+ }
392
+ out := make(map[string]string, len(agg.groupBy))
393
+ for _, l := range agg.groupBy {
394
+ if v := tags[l]; v != "" {
395
+ out[l] = v
396
+ }
397
+ }
398
+ return out
399
+}
400
+
401
+// vmCollapseMetricValue a metric to an int64 quickly; return mv if present for merge path
402
+func vmCollapseMetricValue(m ddsnmp.Metric) (v int64, mv map[string]int64) {
403
+ if len(m.MultiValue) == 0 {
404
+ return m.Value, nil
405
+ }
406
+ var sum int64
407
+ for _, x := range m.MultiValue {
408
+ sum += x
409
+ }
410
+ return sum, m.MultiValue
411
+}
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/collector_vmetrics_test.go
+242
@@ -776,6 +776,248 @@ func TestVirtualMetricsCollector_Collect(t *testing.T) {
776
},
777
},
778
},
779
+
780
+ "group_by per-row composite (in/out)": {
781
+ profileDef: &ddprofiledefinition.ProfileDefinition{
782
+ VirtualMetrics: []ddprofiledefinition.VirtualMetricConfig{
783
+ {
784
+ Name: "ifTrafficPerRow",
785
+ GroupBy: []string{"*"},
786
+ Sources: []ddprofiledefinition.VirtualMetricSourceConfig{
787
+ {Metric: "ifHCInOctets", Table: "ifXTable", As: "in"},
788
+ {Metric: "ifHCOutOctets", Table: "ifXTable", As: "out"},
789
+ },
790
+ ChartMeta: ddprofiledefinition.ChartMeta{
791
+ Description: "Per-row traffic (in/out)",
792
+ Family: "Network/Interface/Traffic",
793
+ Unit: "bit/s",
794
+ },
795
+ },
796
+ },
797
+ },
798
+ collectedMetrics: []ddsnmp.Metric{
799
+ // eth0 row
800
+ {Name: "ifHCInOctets", Value: 1000, IsTable: true, Table: "ifXTable",
801
+ Tags: map[string]string{"interface": "eth0", "ifType": "ethernetCsmacd", "ifIndex": "1"}},
802
+ {Name: "ifHCOutOctets", Value: 2000, IsTable: true, Table: "ifXTable",
803
+ Tags: map[string]string{"interface": "eth0", "ifType": "ethernetCsmacd", "ifIndex": "1"}},
804
+ // lo row
805
+ {Name: "ifHCInOctets", Value: 10, IsTable: true, Table: "ifXTable",
806
+ Tags: map[string]string{"interface": "lo", "ifType": "softwareLoopback", "ifIndex": "2"}},
807
+ {Name: "ifHCOutOctets", Value: 15, IsTable: true, Table: "ifXTable",
808
+ Tags: map[string]string{"interface": "lo", "ifType": "softwareLoopback", "ifIndex": "2"}},
809
+ },
810
+ expected: []ddsnmp.Metric{
811
+ {
812
+ Name: "ifTrafficPerRow",
813
+ IsTable: true,
814
+ Table: "ifXTable",
815
+ Tags: map[string]string{"interface": "eth0", "ifType": "ethernetCsmacd", "ifIndex": "1"},
816
+ MultiValue: map[string]int64{"in": 1000, "out": 2000},
817
+ Description: "Per-row traffic (in/out)",
818
+ Family: "Network/Interface/Traffic",
819
+ Unit: "bit/s",
820
+ },
821
+ {
822
+ Name: "ifTrafficPerRow",
823
+ IsTable: true,
824
+ Table: "ifXTable",
825
+ Tags: map[string]string{"interface": "lo", "ifType": "softwareLoopback", "ifIndex": "2"},
826
+ MultiValue: map[string]int64{"in": 10, "out": 15},
827
+ Description: "Per-row traffic (in/out)",
828
+ Family: "Network/Interface/Traffic",
829
+ Unit: "bit/s",
830
+ },
831
+ },
832
+ },
833
+
834
+ "group_by explicit labels (interface,ifType) merges duplicates": {
835
+ profileDef: &ddprofiledefinition.ProfileDefinition{
836
+ VirtualMetrics: []ddprofiledefinition.VirtualMetricConfig{
837
+ {
838
+ Name: "ifTrafficPerInterface",
839
+ GroupBy: []string{"interface", "ifType"},
840
+ Sources: []ddprofiledefinition.VirtualMetricSourceConfig{
841
+ {Metric: "ifHCInOctets", Table: "ifXTable", As: "in"},
842
+ {Metric: "ifHCOutOctets", Table: "ifXTable", As: "out"},
843
+ },
844
+ },
845
+ },
846
+ },
847
+ collectedMetrics: []ddsnmp.Metric{
848
+ // two rows that share the same (interface,ifType) -> should be summed within the group
849
+ {Name: "ifHCInOctets", Value: 0, IsTable: true, Table: "ifXTable",
850
+ Tags: map[string]string{"interface": "bond0", "ifType": "ethernetCsmacd", "ifIndex": "10"}},
851
+ {Name: "ifHCOutOctets", Value: 50, IsTable: true, Table: "ifXTable",
852
+ Tags: map[string]string{"interface": "bond0", "ifType": "ethernetCsmacd", "ifIndex": "10"}},
853
+ {Name: "ifHCInOctets", Value: 0, IsTable: true, Table: "ifXTable",
854
+ Tags: map[string]string{"interface": "bond0", "ifType": "ethernetCsmacd", "ifIndex": "42"}},
855
+ {Name: "ifHCOutOctets", Value: 20, IsTable: true, Table: "ifXTable",
856
+ Tags: map[string]string{"interface": "bond0", "ifType": "ethernetCsmacd", "ifIndex": "42"}},
857
+ },
858
+ expected: []ddsnmp.Metric{
859
+ {
860
+ Name: "ifTrafficPerInterface",
861
+ IsTable: true,
862
+ Table: "ifXTable",
863
+ Tags: map[string]string{"interface": "bond0", "ifType": "ethernetCsmacd"}, // only group_by labels
864
+ MultiValue: map[string]int64{"in": 0, "out": 70},
865
+ },
866
+ },
867
+ },
868
+
869
+ "group_by single-source per interface (Value path)": {
870
+ profileDef: &ddprofiledefinition.ProfileDefinition{
871
+ VirtualMetrics: []ddprofiledefinition.VirtualMetricConfig{
872
+ {
873
+ Name: "ifErrorsPerInterface",
874
+ GroupBy: []string{"interface"},
875
+ Sources: []ddprofiledefinition.VirtualMetricSourceConfig{
876
+ {Metric: "ifInErrors", Table: "ifTable"},
877
+ },
878
+ ChartMeta: ddprofiledefinition.ChartMeta{
879
+ Description: "Per-interface inbound errors",
880
+ Family: "Network/Interface/Errors",
881
+ Unit: "{error}/s",
882
+ },
883
+ },
884
+ },
885
+ },
886
+ collectedMetrics: []ddsnmp.Metric{
887
+ {Name: "ifInErrors", Value: 5, IsTable: true, Table: "ifTable",
888
+ Tags: map[string]string{"interface": "eth0", "ifIndex": "1"}},
889
+ {Name: "ifInErrors", Value: 7, IsTable: true, Table: "ifTable",
890
+ Tags: map[string]string{"interface": "eth1", "ifIndex": "2"}},
891
+ {Name: "ifInErrors", Value: 3, IsTable: true, Table: "ifTable",
892
+ Tags: map[string]string{"interface": "eth0", "ifIndex": "1"}},
893
+ },
894
+ expected: []ddsnmp.Metric{
895
+ {
896
+ Name: "ifErrorsPerInterface",
897
+ IsTable: true,
898
+ Table: "ifTable",
899
+ Tags: map[string]string{"interface": "eth0"},
900
+ Value: 8, // 5 + 3
901
+ Description: "Per-interface inbound errors",
902
+ Family: "Network/Interface/Errors",
903
+ Unit: "{error}/s",
904
+ },
905
+ {
906
+ Name: "ifErrorsPerInterface",
907
+ IsTable: true,
908
+ Table: "ifTable",
909
+ Tags: map[string]string{"interface": "eth1"},
910
+ Value: 7,
911
+ Description: "Per-interface inbound errors",
912
+ Family: "Network/Interface/Errors",
913
+ Unit: "{error}/s",
914
+ },
915
+ },
916
+ },
917
+
918
+ "group_by per-row with missing dim (partial)": {
919
+ profileDef: &ddprofiledefinition.ProfileDefinition{
920
+ VirtualMetrics: []ddprofiledefinition.VirtualMetricConfig{
921
+ {
922
+ Name: "ifTrafficPerRow",
923
+ GroupBy: []string{"*"},
924
+ Sources: []ddprofiledefinition.VirtualMetricSourceConfig{
925
+ {Metric: "ifHCInOctets", Table: "ifXTable", As: "in"},
926
+ {Metric: "ifHCOutOctets", Table: "ifXTable", As: "out"},
927
+ },
928
+ },
929
+ },
930
+ },
931
+ collectedMetrics: []ddsnmp.Metric{
932
+ // eth0 has only IN
933
+ {Name: "ifHCInOctets", Value: 111, IsTable: true, Table: "ifXTable",
934
+ Tags: map[string]string{"interface": "eth0", "ifType": "ethernetCsmacd", "ifIndex": "1"}},
935
+ // eth1 has only OUT
936
+ {Name: "ifHCOutOctets", Value: 222, IsTable: true, Table: "ifXTable",
937
+ Tags: map[string]string{"interface": "eth1", "ifType": "ethernetCsmacd", "ifIndex": "2"}},
938
+ },
939
+ expected: []ddsnmp.Metric{
940
+ {
941
+ Name: "ifTrafficPerRow",
942
+ IsTable: true,
943
+ Table: "ifXTable",
944
+ Tags: map[string]string{"interface": "eth0", "ifType": "ethernetCsmacd", "ifIndex": "1"},
945
+ MultiValue: map[string]int64{"in": 111}, // out omitted
946
+ },
947
+ {
948
+ Name: "ifTrafficPerRow",
949
+ IsTable: true,
950
+ Table: "ifXTable",
951
+ Tags: map[string]string{"interface": "eth1", "ifType": "ethernetCsmacd", "ifIndex": "2"},
952
+ MultiValue: map[string]int64{"out": 222},
953
+ },
954
+ },
955
+ },
956
+
957
+ "group_by explicit labels but sources span tables (skipped VM)": {
958
+ profileDef: &ddprofiledefinition.ProfileDefinition{
959
+ VirtualMetrics: []ddprofiledefinition.VirtualMetricConfig{
960
+ {
961
+ Name: "invalidGroupedVM",
962
+ GroupBy: []string{"interface"},
963
+ Sources: []ddprofiledefinition.VirtualMetricSourceConfig{
964
+ {Metric: "ifInOctets", Table: "ifTable", As: "in"},
965
+ {Metric: "ifHCInOctets", Table: "ifXTable", As: "in2"},
966
+ },
967
+ },
968
+ },
969
+ },
970
+ collectedMetrics: []ddsnmp.Metric{
971
+ {Name: "ifInOctets", Value: 100, IsTable: true, Table: "ifTable",
972
+ Tags: map[string]string{"interface": "eth0", "ifIndex": "1"}},
973
+ {Name: "ifHCInOctets", Value: 200, IsTable: true, Table: "ifXTable",
974
+ Tags: map[string]string{"interface": "eth0", "ifIndex": "1"}},
975
+ },
976
+ expected: []ddsnmp.Metric{}, // VM skipped during build phase
977
+ },
978
+
979
+ "group_by per-row without index/ifIndex (fallback key)": {
980
+ profileDef: &ddprofiledefinition.ProfileDefinition{
981
+ VirtualMetrics: []ddprofiledefinition.VirtualMetricConfig{
982
+ {
983
+ Name: "ifTrafficPerRow",
984
+ GroupBy: []string{"*"},
985
+ Sources: []ddprofiledefinition.VirtualMetricSourceConfig{
986
+ {Metric: "ifHCInOctets", Table: "ifXTable", As: "in"},
987
+ {Metric: "ifHCOutOctets", Table: "ifXTable", As: "out"},
988
+ },
989
+ },
990
+ },
991
+ },
992
+ collectedMetrics: []ddsnmp.Metric{
993
+ // tags without index/ifIndex; fallback should still build a stable key
994
+ {Name: "ifHCInOctets", Value: 5, IsTable: true, Table: "ifXTable",
995
+ Tags: map[string]string{"interface": "ethA", "ifType": "ethernetCsmacd"}},
996
+ {Name: "ifHCOutOctets", Value: 7, IsTable: true, Table: "ifXTable",
997
+ Tags: map[string]string{"interface": "ethA", "ifType": "ethernetCsmacd"}},
998
+
999
+ {Name: "ifHCInOctets", Value: 1, IsTable: true, Table: "ifXTable",
1000
+ Tags: map[string]string{"interface": "ethB", "ifType": "ethernetCsmacd"}},
1001
+ {Name: "ifHCOutOctets", Value: 2, IsTable: true, Table: "ifXTable",
1002
+ Tags: map[string]string{"interface": "ethB", "ifType": "ethernetCsmacd"}},
1003
+ },
1004
+ expected: []ddsnmp.Metric{
1005
+ {
1006
+ Name: "ifTrafficPerRow",
1007
+ IsTable: true,
1008
+ Table: "ifXTable",
1009
+ Tags: map[string]string{"interface": "ethA", "ifType": "ethernetCsmacd"},
1010
+ MultiValue: map[string]int64{"in": 5, "out": 7},
1011
+ },
1012
+ {
1013
+ Name: "ifTrafficPerRow",
1014
+ IsTable: true,
1015
+ Table: "ifXTable",
1016
+ Tags: map[string]string{"interface": "ethB", "ifType": "ethernetCsmacd"},
1017
+ MultiValue: map[string]int64{"in": 1, "out": 2},
1018
+ },
1019
+ },
1020
+ },
1021
}
1022
1023
for name, tc := range tests {