perf(go.d/ddsnmp): cache group key per aggregator per metric (#20979)
Ilya Mashchenko committed
Sep 14, 2025 at 21:32 UTC
470fe48e9c48c4700572cf0c670306127e885f9f
2 files changed
+200
-58
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/collector_vmetrics.go
+66
-58
@@ -40,6 +40,12 @@ func (p *vmetricsCollector) accumulate(lookup map[vmetricsSourceKey][]vmetricsSi
40
41
v, mv := vmCollapseMetricValue(m)
42
43
+ type gke struct {
44
+ key string
45
+ ok bool
46
+ }
47
+ var gkCache map[*vmetricsAggregator]gke
48
+
49
for _, sink := range sinks {
50
agg := sink.agg
51
if agg == nil {
@@ -50,7 +56,24 @@ func (p *vmetricsCollector) accumulate(lookup map[vmetricsSourceKey][]vmetricsSi
56
agg.metricType = m.MetricType
57
}
58
53
- agg.accumulate(sink, v, mv, m.Tags)
59
+ if !agg.grouped {
60
+ agg.accumulateTotal(sink, v, mv)
61
+ continue
62
+ }
63
+
64
+ if gkCache == nil {
65
+ gkCache = make(map[*vmetricsAggregator]gke, 2)
66
+ }
67
+ entry, found := gkCache[agg]
68
+ if !found {
69
+ k, ok := vmBuildGroupKey(m.Tags, agg)
70
+ entry = gke{key: k, ok: ok}
71
+ gkCache[agg] = entry
72
+ }
73
+ if !entry.ok {
74
+ continue
75
+ }
76
+ agg.accumulateGroupedWithKey(sink, entry.key, v, m.Tags)
77
}
78
}
79
}
@@ -59,7 +82,7 @@ func (p *vmetricsCollector) emit(aggrs []*vmetricsAggregator) []ddsnmp.Metric {
82
// small pre-alloc: 1 total or ~groups count; start conservative
83
out := make([]ddsnmp.Metric, 0, len(aggrs))
84
for _, agg := range aggrs {
62
- if agg.sourceCount == 0 {
85
+ if !agg.hadData {
86
p.log.Debugf("no source metrics found for virtual metric '%s'", agg.config.Name)
87
continue
88
}
@@ -82,15 +105,15 @@ type (
105
perGroup map[string]*vmetricsGroupBucket
106
107
// --- dimensions (composite) ---
85
- dims vmetricsDimSpec
108
+ dimNames []string
109
110
// --- 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
111
+ perDim map[string]int64 // composite total (dim -> sum)
112
+ sum int64 // single-source total
113
+ multiSum map[string]int64 // merged base MultiValue
114
93
- keyBuf strings.Builder
115
+ hadData bool
116
+ keyBuf strings.Builder
117
}
118
119
// vmetricsSourceKey identifies a metric source
@@ -105,13 +128,6 @@ type (
128
dimIdx int16
129
}
130
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
-
131
// per-group accumulator (emitted as one table row)
132
vmetricsGroupBucket struct {
133
vals []int64 // len == dims.count when composite
@@ -121,43 +137,14 @@ type (
137
}
138
)
139
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
-
140
func (agg *vmetricsAggregator) accumulateTotal(sink vmetricsSink, v int64, mv map[string]int64) {
141
+ agg.hadData = true
142
+
143
if sink.dimIdx >= 0 {
144
if agg.perDim == nil {
158
- agg.perDim = make(map[string]int64, agg.dims.count)
145
+ agg.perDim = make(map[string]int64, len(agg.dimNames))
146
}
160
- name := agg.dims.names[sink.dimIdx]
147
+ name := agg.dimNames[sink.dimIdx]
148
agg.perDim[name] += v
149
return
150
}
@@ -174,6 +161,26 @@ func (agg *vmetricsAggregator) accumulateTotal(sink vmetricsSink, v int64, mv ma
161
agg.sum += v
162
}
163
164
+func (agg *vmetricsAggregator) accumulateGroupedWithKey(sink vmetricsSink, gkey string, v int64, tags map[string]string) {
165
+ agg.hadData = true
166
+
167
+ b := agg.perGroup[gkey]
168
+ if b == nil {
169
+ b = &vmetricsGroupBucket{emitTags: vmBuildEmitTags(tags, agg)}
170
+ if len(agg.dimNames) > 0 {
171
+ b.vals = make([]int64, len(agg.dimNames))
172
+ b.seen = make([]bool, len(agg.dimNames))
173
+ }
174
+ agg.perGroup[gkey] = b
175
+ }
176
+ if sink.dimIdx >= 0 {
177
+ b.vals[sink.dimIdx] += v
178
+ b.seen[sink.dimIdx] = true
179
+ } else {
180
+ b.sum += v
181
+ }
182
+}
183
+
184
func (agg *vmetricsAggregator) emitInto(out *[]ddsnmp.Metric) {
185
if agg.grouped {
186
agg.emitGrouped(out)
@@ -195,9 +202,9 @@ func (agg *vmetricsAggregator) emitGrouped(out *[]ddsnmp.Metric) {
202
Table: agg.groupTable,
203
Tags: b.emitTags,
204
}
198
- if agg.dims.count > 0 {
199
- mv := make(map[string]int64, agg.dims.count)
200
- for i, dn := range agg.dims.names {
205
+ if len(agg.dimNames) > 0 {
206
+ mv := make(map[string]int64, len(agg.dimNames))
207
+ for i, dn := range agg.dimNames {
208
if b.seen[i] {
209
mv[dn] = b.vals[i]
210
}
@@ -275,17 +282,18 @@ func (p *vmetricsCollector) buildAggregators(profDef *ddprofiledefinition.Profil
282
return s.As != ""
283
})
284
285
+ var dimsIdxByName map[string]int
286
+
287
if isComposite {
279
- agg.dims.idxByName = make(map[string]int, len(cfg.Sources))
280
- agg.dims.names = make([]string, 0, len(cfg.Sources))
288
+ dimsIdxByName = make(map[string]int, len(cfg.Sources))
289
+ agg.dimNames = make([]string, 0, len(cfg.Sources))
290
for _, s := range cfg.Sources {
291
name := ternary(s.As != "", s.As, s.Metric)
283
- if _, dup := agg.dims.idxByName[name]; !dup {
284
- agg.dims.idxByName[name] = len(agg.dims.names)
285
- agg.dims.names = append(agg.dims.names, name)
292
+ if _, dup := dimsIdxByName[name]; !dup {
293
+ dimsIdxByName[name] = len(agg.dimNames)
294
+ agg.dimNames = append(agg.dimNames, name)
295
}
296
}
288
- agg.dims.count = len(agg.dims.names)
297
}
298
299
// register sinks
@@ -295,7 +303,7 @@ func (p *vmetricsCollector) buildAggregators(profDef *ddprofiledefinition.Profil
303
dimIdx := int16(-1)
304
if isComposite {
305
name := ternary(src.As != "", src.As, src.Metric)
298
- if idx, ok := agg.dims.idxByName[name]; ok {
306
+ if idx, ok := dimsIdxByName[name]; ok {
307
dimIdx = int16(idx)
308
}
309
}
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/collector_vmetrics_test.go
+134
@@ -3,6 +3,7 @@
3
package ddsnmpcollector
4
5
import (
6
+ "strconv"
7
"testing"
8
9
"github.com/stretchr/testify/assert"
@@ -1185,3 +1186,136 @@ func TestVirtualMetricsCollector_Collect(t *testing.T) {
1186
})
1187
}
1188
}
1189
+
1190
+var (
1191
+ benchSinkString string
1192
+ benchSinkBucket *vmetricsGroupBucket
1193
+)
1194
+
1195
+func makeTags(n int) map[string]string {
1196
+ t := make(map[string]string, n)
1197
+ for i := 0; i < n; i++ {
1198
+ t["label"+strconv.Itoa(i)] = "v" + strconv.Itoa(i)
1199
+ }
1200
+ return t
1201
+}
1202
+
1203
+func makeAgg(perRow bool, groupBy []string) *vmetricsAggregator {
1204
+ return &vmetricsAggregator{
1205
+ perRow: perRow,
1206
+ groupBy: groupBy,
1207
+ grouped: perRow || len(groupBy) > 0,
1208
+ }
1209
+}
1210
+
1211
+func Benchmark_vmBuildGroupKey(b *testing.B) {
1212
+ type testCase struct {
1213
+ name string
1214
+ perRow bool
1215
+ groupBy []string
1216
+ tagCount int
1217
+ expectKey bool
1218
+ }
1219
+
1220
+ cases := []testCase{
1221
+ {name: "PerRow_GroupBy2", perRow: true, groupBy: []string{"label0", "label1"}, tagCount: 10, expectKey: true},
1222
+ {name: "PerRow_Fallback_AllTags10", perRow: true, groupBy: nil, tagCount: 10, expectKey: true},
1223
+ {name: "PerRow_Fallback_AllTags30", perRow: true, groupBy: nil, tagCount: 30, expectKey: true},
1224
+ {name: "GroupBy1", perRow: false, groupBy: []string{"label0"}, tagCount: 10, expectKey: true},
1225
+ {name: "GroupBy2", perRow: false, groupBy: []string{"label0", "label1"}, tagCount: 10, expectKey: true},
1226
+ {name: "GroupBy3", perRow: false, groupBy: []string{"label0", "label1", "label2"}, tagCount: 10, expectKey: true},
1227
+ {name: "GroupBy2_MissingLabel", perRow: false, groupBy: []string{"label0", "missing"}, tagCount: 10, expectKey: false},
1228
+ }
1229
+
1230
+ for _, tc := range cases {
1231
+ b.Run(tc.name, func(b *testing.B) {
1232
+ agg := makeAgg(tc.perRow, tc.groupBy)
1233
+ tags := makeTags(tc.tagCount)
1234
+ b.ReportAllocs()
1235
+ b.ResetTimer()
1236
+ for i := 0; i < b.N; i++ {
1237
+ key, ok := vmBuildGroupKey(tags, agg)
1238
+ if ok != tc.expectKey {
1239
+ b.Fatalf("expected ok=%v, got %v", tc.expectKey, ok)
1240
+ }
1241
+ // prevent dead-code elimination
1242
+ if ok {
1243
+ benchSinkString = key
1244
+ }
1245
+ }
1246
+ })
1247
+ }
1248
+}
1249
+
1250
+func Benchmark_CollectorAccumulate_PerRowFallback(b *testing.B) {
1251
+ type tc struct {
1252
+ name string
1253
+ sinks int // number of sinks feeding the same aggregator (simulate composite dims)
1254
+ tagCount int // number of tags to force the sort+join fallback
1255
+ }
1256
+ cases := []tc{
1257
+ {name: "Sinks2_Tags10", sinks: 2, tagCount: 10},
1258
+ {name: "Sinks4_Tags10", sinks: 4, tagCount: 10},
1259
+ {name: "Sinks2_Tags30", sinks: 2, tagCount: 30},
1260
+ }
1261
+
1262
+ for _, c := range cases {
1263
+ b.Run(c.name, func(b *testing.B) {
1264
+ // --- Arrange ---
1265
+ const metricName = "raw_in"
1266
+ const tableName = "ifTable"
1267
+
1268
+ // Aggregator configured as: grouped, PerRow=true, no groupBy (force fallback)
1269
+ agg := &vmetricsAggregator{
1270
+ config: ddprofiledefinition.VirtualMetricConfig{Name: "vm_perrow"},
1271
+ grouped: true,
1272
+ perRow: true,
1273
+ groupBy: nil,
1274
+ groupTable: tableName,
1275
+ perGroup: make(map[string]*vmetricsGroupBucket, 64),
1276
+ dimNames: make([]string, c.sinks),
1277
+ }
1278
+ for i := 0; i < c.sinks; i++ {
1279
+ agg.dimNames[i] = "d" + strconv.Itoa(i)
1280
+ }
1281
+
1282
+ // Build lookup with N sinks pointing to the SAME aggregator.
1283
+ lookup := make(map[vmetricsSourceKey][]vmetricsSink, 1)
1284
+ key := vmetricsSourceKey{metricName: metricName, tableName: tableName}
1285
+ sinks := make([]vmetricsSink, c.sinks)
1286
+ for i := 0; i < c.sinks; i++ {
1287
+ sinks[i] = vmetricsSink{agg: agg, dimIdx: int16(i)}
1288
+ }
1289
+ lookup[key] = sinks
1290
+
1291
+ // One metric instance (like one SNMP sample)
1292
+ tags := makeTags(c.tagCount)
1293
+ m := ddsnmp.Metric{
1294
+ Name: metricName,
1295
+ Table: tableName,
1296
+ Value: 1,
1297
+ MetricType: ddprofiledefinition.ProfileMetricType("counter"),
1298
+ Tags: tags,
1299
+ }
1300
+ collected := []ddsnmp.Metric{m}
1301
+
1302
+ collector := &vmetricsCollector{log: nil} // log unused on accumulate
1303
+
1304
+ // Warm-up once so the bucket exists; we want to measure repeated key-building.
1305
+ collector.accumulate(lookup, collected)
1306
+
1307
+ // --- Measure ---
1308
+ b.ReportAllocs()
1309
+ b.ResetTimer()
1310
+ for i := 0; i < b.N; i++ {
1311
+ collector.accumulate(lookup, collected)
1312
+ }
1313
+
1314
+ // Prevent DCE: observe a bucket.
1315
+ for _, v := range agg.perGroup {
1316
+ benchSinkBucket = v
1317
+ break
1318
+ }
1319
+ })
1320
+ }
1321
+}