@cryptotaxi247 / netdata / commits / 0360d0efb

chore(go.d/ddsnmp): fix table metrics collection (#20492)

Ilya Mashchenko committed Jun 16, 2025 at 15:19 UTC 0360d0efbe030f978bfb1567ad35a328c9f6994a
7 files changed +472 -355
src/go/plugin/go.d/collector/snmp/collect_profiles.go
+40 -38
@@ -24,60 +24,62 @@ func (c *Collector) collectProfiles(mx map[string]int64) error {
24 return err
25 }
26
27 - for _, pm := range pms {
28 - c.collectProfileScalarMetrics(mx, pm)
29 - c.collectProfileTableMetrics(mx, pm)
30 - }
27 + c.collectProfileScalarMetrics(mx, pms)
28 + c.collectProfileTableMetrics(mx, pms)
29
30 return nil
31 }
32
35 -func (c *Collector) collectProfileScalarMetrics(mx map[string]int64, pm *ddsnmpcollector.ProfileMetrics) {
36 - for _, m := range pm.Metrics {
37 - if m.IsTable || m.Name == "" {
38 - continue
39 - }
33 +func (c *Collector) collectProfileScalarMetrics(mx map[string]int64, pms []*ddsnmpcollector.ProfileMetrics) {
34 + for _, pm := range pms {
35 + for _, m := range pm.Metrics {
36 + if m.IsTable || m.Name == "" {
37 + continue
38 + }
39
41 - if !c.seenScalarMetrics[m.Name] {
42 - c.seenScalarMetrics[m.Name] = true
43 - c.addProfileScalarMetricChart(m)
44 - }
40 + if !c.seenScalarMetrics[m.Name] {
41 + c.seenScalarMetrics[m.Name] = true
42 + c.addProfileScalarMetricChart(m)
43 + }
44
46 - if len(m.Mappings) == 0 {
47 - id := fmt.Sprintf("snmp_device_prof_%s", m.Name)
48 - mx[id] = m.Value
49 - } else {
50 - for k, v := range m.Mappings {
51 - id := fmt.Sprintf("snmp_device_prof_%s_%s", m.Name, v)
52 - mx[id] = metrix.Bool(m.Value == k)
45 + if len(m.Mappings) == 0 {
46 + id := fmt.Sprintf("snmp_device_prof_%s", m.Name)
47 + mx[id] = m.Value
48 + } else {
49 + for k, v := range m.Mappings {
50 + id := fmt.Sprintf("snmp_device_prof_%s_%s", m.Name, v)
51 + mx[id] = metrix.Bool(m.Value == k)
52 + }
53 }
54 }
55 }
56 }
57
58 -func (c *Collector) collectProfileTableMetrics(mx map[string]int64, pm *ddsnmpcollector.ProfileMetrics) {
58 +func (c *Collector) collectProfileTableMetrics(mx map[string]int64, pms []*ddsnmpcollector.ProfileMetrics) {
59 seen := make(map[string]bool)
60
61 - for _, m := range pm.Metrics {
62 - if !m.IsTable || m.Name == "" || len(m.Tags) == 0 {
63 - continue
64 - }
61 + for _, pm := range pms {
62 + for _, m := range pm.Metrics {
63 + if !m.IsTable || m.Name == "" || len(m.Tags) == 0 {
64 + continue
65 + }
66
66 - key := tableMetricKey(m)
67 - seen[key] = true
67 + key := tableMetricKey(m)
68 + seen[key] = true
69
69 - if !c.seenTableMetrics[key] {
70 - c.seenTableMetrics[key] = true
71 - c.addProfileTableMetricChart(m)
72 - }
70 + if !c.seenTableMetrics[key] {
71 + c.seenTableMetrics[key] = true
72 + c.addProfileTableMetricChart(m)
73 + }
74
74 - if len(m.Mappings) == 0 {
75 - id := fmt.Sprintf("snmp_device_prof_%s", key)
76 - mx[id] = m.Value
77 - } else {
78 - for k, v := range m.Mappings {
79 - id := fmt.Sprintf("snmp_device_prof_%s_%s", key, v)
80 - mx[id] = metrix.Bool(m.Value == k)
75 + if len(m.Mappings) == 0 {
76 + id := fmt.Sprintf("snmp_device_prof_%s", key)
77 + mx[id] = m.Value
78 + } else {
79 + for k, v := range m.Mappings {
80 + id := fmt.Sprintf("snmp_device_prof_%s_%s", key, v)
81 + mx[id] = metrix.Bool(m.Value == k)
82 + }
83 }
84 }
85 }
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/collect_table.go
+93 -112
@@ -6,6 +6,7 @@ import (
6 "errors"
7 "fmt"
8 "maps"
9 + "sort"
10 "strings"
11
12 "github.com/gosnmp/gosnmp"
@@ -38,34 +39,56 @@ func (c *Collector) walkAllTables(prof *ddsnmp.Profile) ([]tableWalkResult, erro
39 var errs []error
40 var missingOIDs []string
41
41 - doneOids := make(map[string]bool)
42 + // Map to store walked data by table OID
43 + walkedTables := make(map[string]map[string]gosnmp.SnmpPDU)
44
45 + // Map to track which tables need to be walked
46 + tablesToWalk := make(map[string]bool)
47 +
48 + // First pass: identify unique tables to walk
49 for _, cfg := range prof.Definition.Metrics {
44 - if cfg.IsScalar() || cfg.Table.OID == "" || doneOids[cfg.Table.OID] {
50 + if cfg.IsScalar() || cfg.Table.OID == "" {
51 continue
52 }
53
48 - if c.missingOIDs[trimOID(cfg.Table.OID)] {
49 - missingOIDs = append(missingOIDs, cfg.Table.OID)
54 + tableOID := cfg.Table.OID
55 + if c.missingOIDs[trimOID(tableOID)] {
56 + missingOIDs = append(missingOIDs, tableOID)
57 continue
58 }
59
53 - doneOids[cfg.Table.OID] = true
60 + tablesToWalk[tableOID] = true
61 + }
62
55 - // Walk the table
56 - pdus, err := c.snmpWalk(cfg.Table.OID)
63 + // Walk each unique table once
64 + for tableOID := range tablesToWalk {
65 + pdus, err := c.snmpWalk(tableOID)
66 if err != nil {
58 - errs = append(errs, fmt.Errorf("failed to walk table '%s': %w", cfg.Table.Name, err))
67 + errs = append(errs, fmt.Errorf("failed to walk table OID '%s': %w", tableOID, err))
68 continue
69 }
70
71 if len(pdus) > 0 {
63 - results = append(results, tableWalkResult{
64 - tableOID: cfg.Table.OID,
65 - pdus: pdus,
66 - config: cfg,
67 - })
72 + walkedTables[tableOID] = pdus
73 + }
74 + }
75 +
76 + // Second pass: create results for ALL metric configs
77 + for _, cfg := range prof.Definition.Metrics {
78 + if cfg.IsScalar() || cfg.Table.OID == "" {
79 + continue
80 + }
81 +
82 + pdus, ok := walkedTables[cfg.Table.OID]
83 + if !ok {
84 + continue
85 }
86 +
87 + results = append(results, tableWalkResult{
88 + tableOID: cfg.Table.OID,
89 + pdus: pdus,
90 + config: cfg,
91 + })
92 }
93
94 if len(missingOIDs) > 0 {
@@ -118,8 +141,8 @@ func (c *Collector) processTableWalkResults(walkResults []tableWalkResult) ([]Me
141 // Process a single table's data
142 func (c *Collector) processTableData(cfg ddprofiledefinition.MetricsConfig, pdus map[string]gosnmp.SnmpPDU, allWalkedData map[string]map[string]gosnmp.SnmpPDU, tableNameToOID map[string]string) ([]Metric, error) {
143 // Try to use cache if available
121 - if cachedOIDs, cachedTags, ok := c.tableCache.getCachedData(cfg.Table.OID); ok {
122 - metrics, err := c.collectTableWithCache(cfg, cachedOIDs, cachedTags, buildColumnOIDs(cfg))
144 + if cachedIndexes, ok := c.tableCache.getCachedIndexes(cfg.Table.OID); ok {
145 + metrics, err := c.collectTableWithCache(cfg, cachedIndexes, allWalkedData, tableNameToOID)
146 if err == nil {
147 c.log.Debugf("Successfully collected table %s using cache", cfg.Table.Name)
148 return metrics, nil
@@ -139,28 +162,37 @@ func (c *Collector) processTableData(cfg ddprofiledefinition.MetricsConfig, pdus
162 allColumnOIDs = append(allColumnOIDs, oid)
163 }
164
142 - // Group PDUs by row index and build cache structure
143 - rows := make(map[string]map[string]gosnmp.SnmpPDU)
144 - oidCache := make(map[string]map[string]string) // For caching: index -> column OID -> full OID
145 - tagCache := make(map[string]map[string]string) // For caching: index -> tag name -> value
146 -
147 - for oid, pdu := range pdus {
165 + // Extract unique indexes from walked data
166 + indexSet := make(map[string]bool)
167 + for oid := range pdus {
168 for _, columnOID := range allColumnOIDs {
169 if strings.HasPrefix(oid, columnOID+".") {
170 index := strings.TrimPrefix(oid, columnOID+".")
151 -
152 - if rows[index] == nil {
153 - rows[index] = make(map[string]gosnmp.SnmpPDU)
154 - oidCache[index] = make(map[string]string)
155 - tagCache[index] = make(map[string]string)
156 - }
157 - rows[index][columnOID] = pdu
158 - oidCache[index][columnOID] = oid
171 + indexSet[index] = true
172 break
173 }
174 }
175 }
176
177 + // Convert to sorted slice of indexes
178 + indexes := make([]string, 0, len(indexSet))
179 + for index := range indexSet {
180 + indexes = append(indexes, index)
181 + }
182 + sort.Strings(indexes)
183 +
184 + // Cache the table structure (indexes only)
185 + c.tableCache.cacheIndexes(cfg.Table.OID, indexes)
186 + c.log.Debugf("Cached table %s structure with %d rows", cfg.Table.Name, len(indexes))
187 +
188 + // Now process the walked data to create metrics
189 + return c.processTableRows(cfg, indexes, pdus, allWalkedData, tableNameToOID)
190 +}
191 +
192 +func (c *Collector) processTableRows(cfg ddprofiledefinition.MetricsConfig, indexes []string, pdus map[string]gosnmp.SnmpPDU, allWalkedData map[string]map[string]gosnmp.SnmpPDU, tableNameToOID map[string]string) ([]Metric, error) {
193 + columnOIDs := buildColumnOIDs(cfg)
194 + tagColumnOIDs := buildTagColumnOIDs(cfg)
195 +
196 rowStaticTags := make(map[string]string)
197 for _, tag := range cfg.StaticTags {
198 if n, v, _ := strings.Cut(tag, ":"); n != "" && v != "" {
@@ -170,12 +202,13 @@ func (c *Collector) processTableData(cfg ddprofiledefinition.MetricsConfig, pdus
202
203 var metrics []Metric
204
173 - for index, rowPDUs := range rows {
205 + for _, index := range indexes {
206 rowTags := make(map[string]string)
207
176 - // Process tags for this row
208 + // Process same-table tags
209 for columnOID, tagCfg := range tagColumnOIDs {
178 - pdu, ok := rowPDUs[columnOID]
210 + fullOID := columnOID + "." + index
211 + pdu, ok := pdus[fullOID]
212 if !ok {
213 continue
214 }
@@ -188,7 +221,6 @@ func (c *Collector) processTableData(cfg ddprofiledefinition.MetricsConfig, pdus
221
222 for k, v := range tags {
223 rowTags[k] = v
191 - tagCache[index][k] = v
224 }
225 }
226
@@ -199,7 +231,7 @@ func (c *Collector) processTableData(cfg ddprofiledefinition.MetricsConfig, pdus
231 continue
232 }
233
202 - // Skip if it's an index-based tag (handled separately)
234 + // Skip if it's an index-based tag
235 if tagCfg.Index != 0 {
236 continue
237 }
@@ -218,6 +250,7 @@ func (c *Collector) processTableData(cfg ddprofiledefinition.MetricsConfig, pdus
250 continue
251 }
252
253 + // Determine the index to use for lookup
254 lookupIndex := index
255 if len(tagCfg.IndexTransform) > 0 {
256 lookupIndex = applyIndexTransform(index, tagCfg.IndexTransform)
@@ -227,7 +260,7 @@ func (c *Collector) processTableData(cfg ddprofiledefinition.MetricsConfig, pdus
260 }
261 }
262
230 - // Look up the value from the referenced table using the same index
263 + // Look up the value from the referenced table
264 refColumnOID := trimOID(tagCfg.Symbol.OID)
265 refFullOID := refColumnOID + "." + lookupIndex
266
@@ -246,18 +279,16 @@ func (c *Collector) processTableData(cfg ddprofiledefinition.MetricsConfig, pdus
279
280 for k, v := range tags {
281 rowTags[k] = v
249 - tagCache[index][k] = v
282 }
283 }
284
285 // Process index-based tags
286 for _, tagCfg := range cfg.MetricTags {
255 - // Skip if not an index-based tag
287 if tagCfg.Index == 0 {
288 continue
289 }
290
260 - indexValue, ok := getIndexPosition(index, tagCfg.Index)
291 + indexValue, ok := getIndexPosition(index, uint(tagCfg.Index))
292 if !ok {
293 c.log.Debugf("Cannot extract position %d from index %s", tagCfg.Index, index)
294 continue
@@ -265,17 +296,19 @@ func (c *Collector) processTableData(cfg ddprofiledefinition.MetricsConfig, pdus
296
297 tagName := ternary(tagCfg.Tag != "", tagCfg.Tag, fmt.Sprintf("index%d", tagCfg.Index))
298
268 - if v, ok := tagCfg.Mapping[indexValue]; ok {
269 - indexValue = v
299 + if len(tagCfg.Mapping) > 0 {
300 + if mappedValue, ok := tagCfg.Mapping[indexValue]; ok {
301 + indexValue = mappedValue
302 + }
303 }
304
305 rowTags[tagName] = indexValue
273 - tagCache[index][tagName] = indexValue
306 }
307
308 // Process metrics for this row
309 for columnOID, sym := range columnOIDs {
278 - pdu, ok := rowPDUs[columnOID]
310 + fullOID := columnOID + "." + index
311 + pdu, ok := pdus[fullOID]
312 if !ok {
313 continue
314 }
@@ -303,10 +336,6 @@ func (c *Collector) processTableData(cfg ddprofiledefinition.MetricsConfig, pdus
336 }
337 }
338
306 - // Cache the processed data
307 - c.tableCache.cacheData(cfg.Table.OID, oidCache, tagCache)
308 - c.log.Debugf("Cached table %s structure with %d rows", cfg.Table.Name, len(oidCache))
309 -
339 return metrics, nil
340 }
341
@@ -330,18 +359,23 @@ func buildTagColumnOIDs(cfg ddprofiledefinition.MetricsConfig) map[string]ddprof
359
360 func (c *Collector) collectTableWithCache(
361 cfg ddprofiledefinition.MetricsConfig,
333 - cachedOIDs map[string]map[string]string,
334 - cachedTags map[string]map[string]string,
335 - columnOIDs map[string]ddprofiledefinition.SymbolConfig,
362 + cachedIndexes []string,
363 + allWalkedData map[string]map[string]gosnmp.SnmpPDU,
364 + tableNameToOID map[string]string,
365 ) ([]Metric, error) {
337 - var oidsToGet []string
366 + // Build list of OIDs to GET based on cached indexes
367 + columnOIDs := buildColumnOIDs(cfg)
368 + tagColumnOIDs := buildTagColumnOIDs(cfg)
369
339 - for _, columns := range cachedOIDs {
340 - for columnOID, fullOID := range columns {
341 - // Only GET metric columns, tags are cached
342 - if _, isMetric := columnOIDs[columnOID]; isMetric {
343 - oidsToGet = append(oidsToGet, fullOID)
344 - }
370 + var oidsToGet []string
371 + for _, index := range cachedIndexes {
372 + // Get metric columns
373 + for columnOID := range columnOIDs {
374 + oidsToGet = append(oidsToGet, columnOID+"."+index)
375 + }
376 + // Get tag columns (same table only)
377 + for columnOID := range tagColumnOIDs {
378 + oidsToGet = append(oidsToGet, columnOID+"."+index)
379 }
380 }
381
@@ -354,65 +388,12 @@ func (c *Collector) collectTableWithCache(
388 return nil, fmt.Errorf("failed to get cached OIDs: %w", err)
389 }
390
357 - if len(pdus) < len(oidsToGet)/2 { // If we got less than half, probably table structure changed
391 + if len(pdus) < len(oidsToGet)/2 { // If we got less than half, table structure probably changed
392 return nil, fmt.Errorf("table structure may have changed, got %d/%d PDUs", len(pdus), len(oidsToGet))
393 }
394
361 - rowStaticTags := make(map[string]string)
362 -
363 - for _, tag := range cfg.StaticTags {
364 - if n, v, _ := strings.Cut(tag, ":"); n != "" && v != "" {
365 - rowStaticTags[n] = v
366 - }
367 - }
368 -
369 - var metrics []Metric
370 -
371 - for index, columns := range cachedOIDs {
372 - rowTags := make(map[string]string)
373 -
374 - if tags, ok := cachedTags[index]; ok {
375 - for k, v := range tags {
376 - rowTags[k] = v
377 - }
378 - }
379 -
380 - for columnOID, fullOID := range columns {
381 - sym, isMetric := columnOIDs[columnOID]
382 - if !isMetric {
383 - continue
384 - }
385 -
386 - pdu, ok := pdus[trimOID(fullOID)]
387 - if !ok {
388 - c.log.Debugf("Missing PDU for cached OID %s", fullOID)
389 - continue
390 - }
391 -
392 - value, err := processSymbolValue(sym, pdu)
393 - if err != nil {
394 - c.log.Debugf("Error processing value for %s: %v", sym.Name, err)
395 - continue
396 - }
397 -
398 - metric := Metric{
399 - Name: sym.Name,
400 - Value: value,
401 - StaticTags: ternary(len(rowStaticTags) > 0, rowStaticTags, nil),
402 - Tags: ternary(len(rowTags) > 0, rowTags, nil),
403 - Unit: sym.Unit,
404 - Description: sym.Description,
405 - MetricType: getMetricType(sym, pdu),
406 - Family: sym.Family,
407 - Mappings: convSymMappingToNumeric(sym),
408 - IsTable: true,
409 - }
410 -
411 - metrics = append(metrics, metric)
412 - }
413 - }
414 -
415 - return metrics, nil
395 + // Process the rows using the same logic
396 + return c.processTableRows(cfg, cachedIndexes, pdus, allWalkedData, tableNameToOID)
397 }
398
399 func processTableMetricTagValue(cfg ddprofiledefinition.MetricTagConfig, pdu gosnmp.SnmpPDU) (map[string]string, error) {
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/collector.go
+1 -1
@@ -100,7 +100,7 @@ func (c *Collector) Collect() ([]*ProfileMetrics, error) {
100 }
101
102 c.updateMetricFamily(metrics)
103 - cleanTags(metrics)
103 + cleanMetrics(metrics)
104
105 return metrics, nil
106 }
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/collector_test.go
+191
@@ -3379,6 +3379,197 @@ func TestCollector_Collect(t *testing.T) {
3379 }
3380 }
3381
3382 +//func TestCollector_TableCache(t *testing.T) {
3383 +// ctrl := gomock.NewController(t)
3384 +// defer ctrl.Finish()
3385 +//
3386 +// mockHandler := snmpmock.NewMockHandler(ctrl)
3387 +// mockHandler.EXPECT().MaxOids().Return(10).AnyTimes()
3388 +// mockHandler.EXPECT().Version().Return(gosnmp.Version2c).AnyTimes()
3389 +//
3390 +// // First collection - expect table walk
3391 +// mockHandler.EXPECT().BulkWalkAll("1.3.6.1.2.1.2.2").Return(
3392 +// []gosnmp.SnmpPDU{
3393 +// // Row 1
3394 +// {Name: "1.3.6.1.2.1.2.2.1.2.1", Type: gosnmp.OctetString, Value: []byte("eth0")},
3395 +// {Name: "1.3.6.1.2.1.2.2.1.3.1", Type: gosnmp.Integer, Value: 6},
3396 +// {Name: "1.3.6.1.2.1.2.2.1.8.1", Type: gosnmp.Integer, Value: 1},
3397 +// {Name: "1.3.6.1.2.1.2.2.1.10.1", Type: gosnmp.Counter32, Value: uint(1000)},
3398 +// {Name: "1.3.6.1.2.1.2.2.1.16.1", Type: gosnmp.Counter32, Value: uint(2000)},
3399 +// // Row 2
3400 +// {Name: "1.3.6.1.2.1.2.2.1.2.2", Type: gosnmp.OctetString, Value: []byte("lo0")},
3401 +// {Name: "1.3.6.1.2.1.2.2.1.3.2", Type: gosnmp.Integer, Value: 1},
3402 +// {Name: "1.3.6.1.2.1.2.2.1.8.2", Type: gosnmp.Integer, Value: 1},
3403 +// {Name: "1.3.6.1.2.1.2.2.1.10.2", Type: gosnmp.Counter32, Value: uint(500)},
3404 +// {Name: "1.3.6.1.2.1.2.2.1.16.2", Type: gosnmp.Counter32, Value: uint(500)},
3405 +// }, nil,
3406 +// ).Times(1) // Walk only once
3407 +//
3408 +// // Second collection - expect GETs for each config
3409 +// // First config GETs its columns
3410 +// mockHandler.EXPECT().Get(gomock.InAnyOrder([]string{
3411 +// "1.3.6.1.2.1.2.2.1.10.1", // ifInOctets.1
3412 +// "1.3.6.1.2.1.2.2.1.2.1", // ifDescr.1 (tag)
3413 +// "1.3.6.1.2.1.2.2.1.10.2", // ifInOctets.2
3414 +// "1.3.6.1.2.1.2.2.1.2.2", // ifDescr.2 (tag)
3415 +// })).Return(
3416 +// &gosnmp.SnmpPacket{
3417 +// Variables: []gosnmp.SnmpPDU{
3418 +// {Name: "1.3.6.1.2.1.2.2.1.10.1", Type: gosnmp.Counter32, Value: uint(1500)},
3419 +// {Name: "1.3.6.1.2.1.2.2.1.2.1", Type: gosnmp.OctetString, Value: []byte("eth0")},
3420 +// {Name: "1.3.6.1.2.1.2.2.1.10.2", Type: gosnmp.Counter32, Value: uint(600)},
3421 +// {Name: "1.3.6.1.2.1.2.2.1.2.2", Type: gosnmp.OctetString, Value: []byte("lo0")},
3422 +// },
3423 +// }, nil,
3424 +// ).Times(1)
3425 +//
3426 +// // Second config GETs its columns
3427 +// mockHandler.EXPECT().Get(gomock.InAnyOrder([]string{
3428 +// "1.3.6.1.2.1.2.2.1.16.1", // ifOutOctets.1
3429 +// "1.3.6.1.2.1.2.2.1.3.1", // ifType.1 (tag)
3430 +// "1.3.6.1.2.1.2.2.1.16.2", // ifOutOctets.2
3431 +// "1.3.6.1.2.1.2.2.1.3.2", // ifType.2 (tag)
3432 +// })).Return(
3433 +// &gosnmp.SnmpPacket{
3434 +// Variables: []gosnmp.SnmpPDU{
3435 +// {Name: "1.3.6.1.2.1.2.2.1.16.1", Type: gosnmp.Counter32, Value: uint(2500)},
3436 +// {Name: "1.3.6.1.2.1.2.2.1.3.1", Type: gosnmp.Integer, Value: 6},
3437 +// {Name: "1.3.6.1.2.1.2.2.1.16.2", Type: gosnmp.Counter32, Value: uint(600)},
3438 +// {Name: "1.3.6.1.2.1.2.2.1.3.2", Type: gosnmp.Integer, Value: 1},
3439 +// },
3440 +// }, nil,
3441 +// ).Times(1)
3442 +//
3443 +// // Third config GETs its columns
3444 +// mockHandler.EXPECT().Get(gomock.InAnyOrder([]string{
3445 +// "1.3.6.1.2.1.2.2.1.8.1", // ifOperStatus.1
3446 +// "1.3.6.1.2.1.2.2.1.8.2", // ifOperStatus.2
3447 +// })).Return(
3448 +// &gosnmp.SnmpPacket{
3449 +// Variables: []gosnmp.SnmpPDU{
3450 +// {Name: "1.3.6.1.2.1.2.2.1.8.1", Type: gosnmp.Integer, Value: 1},
3451 +// {Name: "1.3.6.1.2.1.2.2.1.8.2", Type: gosnmp.Integer, Value: 1},
3452 +// },
3453 +// }, nil,
3454 +// ).Times(1)
3455 +//
3456 +// // Create profile with multiple configs for same table
3457 +// profile := &ddsnmp.Profile{
3458 +// SourceFile: "test-profile.yaml",
3459 +// Definition: &ddprofiledefinition.ProfileDefinition{
3460 +// Metrics: []ddprofiledefinition.MetricsConfig{
3461 +// {
3462 +// MIB: "IF-MIB",
3463 +// Table: ddprofiledefinition.SymbolConfig{
3464 +// OID: "1.3.6.1.2.1.2.2",
3465 +// Name: "ifTable",
3466 +// },
3467 +// Symbols: []ddprofiledefinition.SymbolConfig{
3468 +// {
3469 +// OID: "1.3.6.1.2.1.2.2.1.10",
3470 +// Name: "ifInOctets",
3471 +// },
3472 +// },
3473 +// MetricTags: []ddprofiledefinition.MetricTagConfig{
3474 +// {
3475 +// Tag: "interface",
3476 +// Symbol: ddprofiledefinition.SymbolConfigCompat{
3477 +// OID: "1.3.6.1.2.1.2.2.1.2",
3478 +// Name: "ifDescr",
3479 +// },
3480 +// },
3481 +// },
3482 +// },
3483 +// {
3484 +// MIB: "IF-MIB",
3485 +// Table: ddprofiledefinition.SymbolConfig{
3486 +// OID: "1.3.6.1.2.1.2.2",
3487 +// Name: "ifTable",
3488 +// },
3489 +// Symbols: []ddprofiledefinition.SymbolConfig{
3490 +// {
3491 +// OID: "1.3.6.1.2.1.2.2.1.16",
3492 +// Name: "ifOutOctets",
3493 +// },
3494 +// },
3495 +// MetricTags: []ddprofiledefinition.MetricTagConfig{
3496 +// {
3497 +// Tag: "if_type",
3498 +// Symbol: ddprofiledefinition.SymbolConfigCompat{
3499 +// OID: "1.3.6.1.2.1.2.2.1.3",
3500 +// Name: "ifType",
3501 +// },
3502 +// Mapping: map[string]string{
3503 +// "6": "ethernet",
3504 +// "1": "other",
3505 +// },
3506 +// },
3507 +// },
3508 +// },
3509 +// {
3510 +// MIB: "IF-MIB",
3511 +// Table: ddprofiledefinition.SymbolConfig{
3512 +// OID: "1.3.6.1.2.1.2.2",
3513 +// Name: "ifTable",
3514 +// },
3515 +// Symbols: []ddprofiledefinition.SymbolConfig{
3516 +// {
3517 +// OID: "1.3.6.1.2.1.2.2.1.8",
3518 +// Name: "ifOperStatus",
3519 +// Mapping: map[string]string{
3520 +// "1": "up",
3521 +// "2": "down",
3522 +// },
3523 +// },
3524 +// },
3525 +// StaticTags: []string{"source:cache_test"},
3526 +// },
3527 +// },
3528 +// },
3529 +// }
3530 +//
3531 +// // Create collector with cache enabled
3532 +// collector := New(mockHandler, []*ddsnmp.Profile{profile}, logger.New())
3533 +// collector.doTableMetrics = true
3534 +// collector.tableCache.setTTL(5*time.Minute, 0.1)
3535 +//
3536 +// // First collection - should walk tables
3537 +// result1, err := collector.Collect()
3538 +// require.NoError(t, err)
3539 +// require.Len(t, result1, 1)
3540 +// require.Len(t, result1[0].Metrics, 6) // 2 rows x 3 metrics
3541 +//
3542 +// // Verify first collection metrics
3543 +// metrics1 := result1[0].Metrics
3544 +// assert.Contains(t, metrics1, Metric{
3545 +// Name: "ifInOctets",
3546 +// Value: 1000,
3547 +// Tags: map[string]string{"interface": "eth0"},
3548 +// MetricType: "rate",
3549 +// IsTable: true,
3550 +// })
3551 +//
3552 +// // Second collection - should use cache
3553 +// result2, err := collector.Collect()
3554 +// require.NoError(t, err)
3555 +// require.Len(t, result2, 1)
3556 +// require.Len(t, result2[0].Metrics, 6) // 2 rows x 3 metrics
3557 +//
3558 +// // Verify second collection has updated values from cache
3559 +// metrics2 := result2[0].Metrics
3560 +// var foundUpdatedMetric bool
3561 +// for _, m := range metrics2 {
3562 +// if m.Name == "ifInOctets" && m.Tags["interface"] == "eth0" {
3563 +// assert.Equal(t, int64(1500), m.Value) // Updated value from cache
3564 +// foundUpdatedMetric = true
3565 +// }
3566 +// }
3567 +// assert.True(t, foundUpdatedMetric, "Should find updated metric from cache")
3568 +//
3569 +// // Verify all expectations were met
3570 +// ctrl.Finish()
3571 +//}
3572 +
3573 func mustCompileRegex(pattern string) *regexp.Regexp {
3574 re, err := regexp.Compile(pattern)
3575 if err != nil {
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/table_cache.go
+28 -110
@@ -9,14 +9,14 @@ import (
9 )
10
11 // Table Cache Overview:
12 -// The table cache converts repeated SNMP walks into efficient GET operations.
13 -// - First collection: Full walk, cache structure and tags
14 -// - Subsequent collections: GET metrics only, use cached tags
15 -// - Tables with dependencies expire together to maintain consistency
12 +// The table cache stores table structure (which rows exist) to convert repeated
13 +// SNMP walks into efficient GET operations.
14 +// - First collection: Full walk, cache row indexes
15 +// - Subsequent collections: GET only the columns needed for each MetricsConfig
16
17 type tableCache struct {
18 - // Table OID -> row index -> column OID -> full OID
19 - tables map[string]map[string]map[string]string
18 + // Table OID -> list of indexes (rows) that exist
19 + tableIndexes map[string][]string
20
21 // Table OID -> when cached
22 timestamps map[string]time.Time
@@ -24,11 +24,7 @@ type tableCache struct {
24 // Table OID -> specific TTL for this table (with jitter applied)
25 tableTTLs map[string]time.Duration
26
27 - // Table OID -> tag values (index -> tag name -> value)
28 - tagValues map[string]map[string]map[string]string
29 -
27 // Table OID -> list of dependent table OIDs (bidirectional)
31 - // If table A depends on table B, both A->B and B->A are stored
28 tableDeps map[string]map[string]bool
29
30 baseTTL time.Duration
@@ -39,14 +35,13 @@ type tableCache struct {
35
36 func newTableCache(baseTTL time.Duration, jitterPct float64) *tableCache {
37 return &tableCache{
42 - tables: make(map[string]map[string]map[string]string),
43 - timestamps: make(map[string]time.Time),
44 - tableTTLs: make(map[string]time.Duration),
45 - tagValues: make(map[string]map[string]map[string]string),
46 - tableDeps: make(map[string]map[string]bool),
47 - baseTTL: baseTTL,
48 - jitterPct: jitterPct,
49 - rng: rand.New(rand.NewSource(time.Now().UnixNano())),
38 + tableIndexes: make(map[string][]string),
39 + timestamps: make(map[string]time.Time),
40 + tableTTLs: make(map[string]time.Duration),
41 + tableDeps: make(map[string]map[string]bool),
42 + baseTTL: baseTTL,
43 + jitterPct: jitterPct,
44 + rng: rand.New(rand.NewSource(time.Now().UnixNano())),
45 }
46 }
47
@@ -61,34 +56,33 @@ func (tc *tableCache) calculateTableTTL() time.Duration {
56 return time.Duration(base * multiplier)
57 }
58
64 -func (tc *tableCache) getCachedData(tableOID string) (oids map[string]map[string]string, tags map[string]map[string]string, found bool) {
59 +func (tc *tableCache) getCachedIndexes(tableOID string) ([]string, bool) {
60 tc.mu.RLock()
61 defer tc.mu.RUnlock()
62
63 if tc.baseTTL == 0 {
69 - return nil, nil, false
64 + return nil, false
65 }
66
67 timestamp, ok := tc.timestamps[tableOID]
68 if !ok {
74 - return nil, nil, false
69 + return nil, false
70 }
71
72 ttl, ok := tc.tableTTLs[tableOID]
73 if !ok || time.Since(timestamp) > ttl {
79 - return nil, nil, false
74 + return nil, false
75 }
76
82 - oids = tc.tables[tableOID]
83 - tags = tc.tagValues[tableOID]
84 - return oids, tags, true
77 + indexes := tc.tableIndexes[tableOID]
78 + return indexes, true
79 }
80
87 -func (tc *tableCache) cacheData(tableOID string, oidMap map[string]map[string]string, tagValues map[string]map[string]string) {
88 - tc.cacheDataWithDeps(tableOID, oidMap, tagValues, nil)
81 +func (tc *tableCache) cacheIndexes(tableOID string, indexes []string) {
82 + tc.cacheIndexesWithDeps(tableOID, indexes, nil)
83 }
84
91 -func (tc *tableCache) cacheDataWithDeps(tableOID string, oidMap map[string]map[string]string, tagValues map[string]map[string]string, dependencies []string) {
85 +func (tc *tableCache) cacheIndexesWithDeps(tableOID string, indexes []string, dependencies []string) {
86 tc.mu.Lock()
87 defer tc.mu.Unlock()
88
@@ -96,27 +90,11 @@ func (tc *tableCache) cacheDataWithDeps(tableOID string, oidMap map[string]map[s
90 return
91 }
92
99 - // Deep copy the maps to avoid reference issues
100 - oidsCopy := make(map[string]map[string]string, len(oidMap))
101 - for index, columns := range oidMap {
102 - columnsCopy := make(map[string]string, len(columns))
103 - for colOID, fullOID := range columns {
104 - columnsCopy[colOID] = fullOID
105 - }
106 - oidsCopy[index] = columnsCopy
107 - }
108 -
109 - tagsCopy := make(map[string]map[string]string, len(tagValues))
110 - for index, tags := range tagValues {
111 - tagCopy := make(map[string]string, len(tags))
112 - for name, value := range tags {
113 - tagCopy[name] = value
114 - }
115 - tagsCopy[index] = tagCopy
116 - }
93 + // Deep copy the indexes
94 + indexesCopy := make([]string, len(indexes))
95 + copy(indexesCopy, indexes)
96
118 - tc.tables[tableOID] = oidsCopy
119 - tc.tagValues[tableOID] = tagsCopy
97 + tc.tableIndexes[tableOID] = indexesCopy
98 tc.timestamps[tableOID] = time.Now()
99 tc.tableTTLs[tableOID] = tc.calculateTableTTL()
100
@@ -164,10 +142,9 @@ func (tc *tableCache) clearExpired() []string {
142
143 // Clear all expired tables
144 for tableOID := range expiredTables {
167 - delete(tc.tables, tableOID)
145 + delete(tc.tableIndexes, tableOID)
146 delete(tc.timestamps, tableOID)
147 delete(tc.tableTTLs, tableOID)
170 - delete(tc.tagValues, tableOID)
148
149 // Clean up dependencies
150 if deps, ok := tc.tableDeps[tableOID]; ok {
@@ -198,68 +175,9 @@ func (tc *tableCache) setTTL(baseTTL time.Duration, jitterPct float64) {
175
176 if baseTTL == 0 {
177 // Clear cache if caching is disabled
201 - tc.tables = make(map[string]map[string]map[string]string)
178 + tc.tableIndexes = make(map[string][]string)
179 tc.timestamps = make(map[string]time.Time)
180 tc.tableTTLs = make(map[string]time.Duration)
204 - tc.tagValues = make(map[string]map[string]map[string]string)
181 tc.tableDeps = make(map[string]map[string]bool)
182 }
183 }
208 -
209 -// Helper method to check if a group of tables is cached
210 -// All tables must be cached and not expired
211 -func (tc *tableCache) areTablesCached(tableOIDs []string) bool {
212 - tc.mu.RLock()
213 - defer tc.mu.RUnlock()
214 -
215 - if tc.baseTTL == 0 {
216 - return false
217 - }
218 -
219 - now := time.Now()
220 - for _, tableOID := range tableOIDs {
221 - timestamp, ok := tc.timestamps[tableOID]
222 - if !ok {
223 - return false
224 - }
225 -
226 - ttl, ok := tc.tableTTLs[tableOID]
227 - if !ok || now.Sub(timestamp) > ttl {
228 - return false
229 - }
230 - }
231 -
232 - return true
233 -}
234 -
235 -func (tc *tableCache) stats() (tables int, withDeps int, totalDeps int) {
236 - tc.mu.RLock()
237 - defer tc.mu.RUnlock()
238 -
239 - tables = len(tc.tables)
240 -
241 - for _, deps := range tc.tableDeps {
242 - if len(deps) > 0 {
243 - withDeps++
244 - totalDeps += len(deps)
245 - }
246 - }
247 -
248 - return tables, withDeps, totalDeps
249 -}
250 -
251 -func (tc *tableCache) getDependencies(tableOID string) []string {
252 - tc.mu.RLock()
253 - defer tc.mu.RUnlock()
254 -
255 - deps, ok := tc.tableDeps[tableOID]
256 - if !ok {
257 - return nil
258 - }
259 -
260 - result := make([]string, 0, len(deps))
261 - for dep := range deps {
262 - result = append(result, dep)
263 - }
264 - return result
265 -}
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/table_cache_test.go
+110 -89
@@ -23,35 +23,21 @@ func TestTableCache(t *testing.T) {
23
24 // Test data
25 tableOID := "1.3.6.1.2.1.2.2"
26 - oidMap := map[string]map[string]string{
27 - "1": {
28 - "1.3.6.1.2.1.2.2.1.2": "1.3.6.1.2.1.2.2.1.2.1",
29 - "1.3.6.1.2.1.2.2.1.10": "1.3.6.1.2.1.2.2.1.10.1",
30 - },
31 - "2": {
32 - "1.3.6.1.2.1.2.2.1.2": "1.3.6.1.2.1.2.2.1.2.2",
33 - "1.3.6.1.2.1.2.2.1.10": "1.3.6.1.2.1.2.2.1.10.2",
34 - },
35 - }
36 - tagValues := map[string]map[string]string{
37 - "1": {"interface": "eth0"},
38 - "2": {"interface": "eth1"},
39 - }
26 + indexes := []string{"1", "2", "3"}
27
28 // Cache data
42 - cache.cacheData(tableOID, oidMap, tagValues)
29 + cache.cacheIndexes(tableOID, indexes)
30
31 // Retrieve cached data - should work
45 - cachedOIDs, cachedTags, found := cache.getCachedData(tableOID)
32 + cachedIndexes, found := cache.getCachedIndexes(tableOID)
33 assert.True(t, found)
47 - assert.Equal(t, oidMap, cachedOIDs)
48 - assert.Equal(t, tagValues, cachedTags)
34 + assert.Equal(t, indexes, cachedIndexes)
35
36 // Wait for expiration (considering jitter)
37 time.Sleep(150 * time.Millisecond)
38
39 // Should be expired now
54 - _, _, found = cache.getCachedData(tableOID)
40 + _, found = cache.getCachedIndexes(tableOID)
41 assert.False(t, found)
42
43 // Clean expired entries
@@ -59,10 +45,9 @@ func TestTableCache(t *testing.T) {
45 assert.Contains(t, expired, tableOID)
46
47 // Cache should be empty now
62 - assert.Empty(t, cache.tables)
48 + assert.Empty(t, cache.tableIndexes)
49 assert.Empty(t, cache.timestamps)
50 assert.Empty(t, cache.tableTTLs)
65 - assert.Empty(t, cache.tagValues)
51 })
52 }
53 }
@@ -75,31 +60,17 @@ func TestTableCacheDependencies(t *testing.T) {
60 table2OID := "1.3.6.1.2.1.31.1.1" // ifXTable
61 table3OID := "1.3.6.1.4.1.9.9.276" // cieIfInterfaceTable
62
78 - oidMap1 := map[string]map[string]string{
79 - "1": {"1.3.6.1.2.1.2.2.1.10": "1.3.6.1.2.1.2.2.1.10.1"},
80 - }
81 - tagValues1 := map[string]map[string]string{
82 - "1": {"interface": "eth0"},
83 - }
84 -
85 - oidMap2 := map[string]map[string]string{
86 - "1": {"1.3.6.1.2.1.31.1.1.1.1": "1.3.6.1.2.1.31.1.1.1.1.1"},
87 - }
88 - tagValues2 := map[string]map[string]string{
89 - "1": {"ifname": "GigabitEthernet0/1"},
90 - }
91 -
92 - oidMap3 := map[string]map[string]string{
93 - "1": {"1.3.6.1.4.1.9.9.276.1.1": "1.3.6.1.4.1.9.9.276.1.1.1"},
94 - }
63 + indexes1 := []string{"1", "2"}
64 + indexes2 := []string{"1", "2"}
65 + indexes3 := []string{"1", "2"}
66
67 // Cache tables with dependencies
68 // table1 and table2 depend on each other
98 - cache.cacheDataWithDeps(table1OID, oidMap1, tagValues1, []string{table2OID})
99 - cache.cacheDataWithDeps(table2OID, oidMap2, tagValues2, []string{table1OID})
69 + cache.cacheIndexesWithDeps(table1OID, indexes1, []string{table2OID})
70 + cache.cacheIndexesWithDeps(table2OID, indexes2, []string{table1OID})
71
72 // table3 depends on table2
102 - cache.cacheDataWithDeps(table3OID, oidMap3, nil, []string{table2OID})
73 + cache.cacheIndexesWithDeps(table3OID, indexes3, []string{table2OID})
74
75 // All tables should be cached
76 assert.True(t, cache.areTablesCached([]string{table1OID, table2OID, table3OID}))
@@ -134,7 +105,7 @@ func TestTableCacheDependencies(t *testing.T) {
105 assert.Contains(t, expired, table3OID)
106
107 // Cache should be empty
137 - assert.Empty(t, cache.tables)
108 + assert.Empty(t, cache.tableIndexes)
109 assert.Empty(t, cache.tableDeps)
110 }
111
@@ -147,13 +118,13 @@ func TestTableCacheDependenciesCascade(t *testing.T) {
118 tableC := "1.3.6.1.2.1.3"
119 tableD := "1.3.6.1.2.1.4"
120
150 - data := map[string]map[string]string{"1": {"col": "val"}}
121 + indexes := []string{"1"}
122
123 // Cache with chain dependencies
153 - cache.cacheDataWithDeps(tableA, data, nil, []string{tableB})
154 - cache.cacheDataWithDeps(tableB, data, nil, []string{tableA, tableC})
155 - cache.cacheDataWithDeps(tableC, data, nil, []string{tableB, tableD})
156 - cache.cacheDataWithDeps(tableD, data, nil, []string{tableC})
124 + cache.cacheIndexesWithDeps(tableA, indexes, []string{tableB})
125 + cache.cacheIndexesWithDeps(tableB, indexes, []string{tableA, tableC})
126 + cache.cacheIndexesWithDeps(tableC, indexes, []string{tableB, tableD})
127 + cache.cacheIndexesWithDeps(tableD, indexes, []string{tableC})
128
129 // All should be cached
130 assert.True(t, cache.areTablesCached([]string{tableA, tableB, tableC, tableD}))
@@ -182,20 +153,23 @@ func TestTableCacheMixedDependencies(t *testing.T) {
153 // Table without deps
154 table3 := "1.3.6.1.2.1.3"
155
185 - data := map[string]map[string]string{"1": {"col": "val"}}
156 + indexes := []string{"1", "2", "3"}
157
158 // Cache tables
188 - cache.cacheDataWithDeps(table1, data, nil, []string{table2})
189 - cache.cacheDataWithDeps(table2, data, nil, []string{table1})
190 - cache.cacheData(table3, data, nil) // No dependencies
159 + cache.cacheIndexesWithDeps(table1, indexes, []string{table2})
160 + cache.cacheIndexesWithDeps(table2, indexes, []string{table1})
161 + cache.cacheIndexes(table3, indexes) // No dependencies
162
163 // All should be cached
193 - _, _, found1 := cache.getCachedData(table1)
194 - _, _, found2 := cache.getCachedData(table2)
195 - _, _, found3 := cache.getCachedData(table3)
164 + indexes1, found1 := cache.getCachedIndexes(table1)
165 + indexes2, found2 := cache.getCachedIndexes(table2)
166 + indexes3, found3 := cache.getCachedIndexes(table3)
167 assert.True(t, found1)
168 assert.True(t, found2)
169 assert.True(t, found3)
170 + assert.Equal(t, indexes, indexes1)
171 + assert.Equal(t, indexes, indexes2)
172 + assert.Equal(t, indexes, indexes3)
173
174 // Wait for expiration
175 time.Sleep(120 * time.Millisecond)
@@ -235,57 +209,48 @@ func TestTableCacheDisabled(t *testing.T) {
209 cache := newTableCache(0, 0) // Disabled cache
210
211 tableOID := "1.3.6.1.2.1.2.2"
238 - oidMap := map[string]map[string]string{
239 - "1": {"1.3.6.1.2.1.2.2.1.2": "1.3.6.1.2.1.2.2.1.2.1"},
240 - }
241 - tagValues := map[string]map[string]string{
242 - "1": {"interface": "eth0"},
243 - }
212 + indexes := []string{"1", "2", "3"}
213
214 // Try to cache data
246 - cache.cacheData(tableOID, oidMap, tagValues)
215 + cache.cacheIndexes(tableOID, indexes)
216
217 // Should not find anything
249 - _, _, found := cache.getCachedData(tableOID)
218 + _, found := cache.getCachedIndexes(tableOID)
219 assert.False(t, found)
220
221 // Try to cache with dependencies
253 - cache.cacheDataWithDeps(tableOID, oidMap, tagValues, []string{"other.table"})
222 + cache.cacheIndexesWithDeps(tableOID, indexes, []string{"other.table"})
223
224 // Should not find anything
256 - _, _, found = cache.getCachedData(tableOID)
225 + _, found = cache.getCachedIndexes(tableOID)
226 assert.False(t, found)
227 assert.False(t, cache.areTablesCached([]string{tableOID}))
228
229 // Cache should remain empty
261 - assert.Empty(t, cache.tables)
230 + assert.Empty(t, cache.tableIndexes)
231 }
232
233 func TestTableCacheDeepCopy(t *testing.T) {
234 cache := newTableCache(1*time.Hour, 0)
235
236 // Original data
268 - oidMap := map[string]map[string]string{
269 - "1": {"col1": "1.2.3.4.1"},
270 - }
271 - tagValues := map[string]map[string]string{
272 - "1": {"tag1": "value1"},
273 - }
237 + indexes := []string{"1", "2", "3"}
238
239 // Cache the data
276 - cache.cacheData("table1", oidMap, tagValues)
240 + cache.cacheIndexes("table1", indexes)
241
278 - // Modify original maps
279 - oidMap["1"]["col2"] = "should not appear"
280 - tagValues["1"]["tag2"] = "should not appear"
242 + // Modify original slice
243 + indexes[0] = "999"
244 + indexes = append(indexes, "4")
245
246 // Retrieve cached data
283 - cachedOIDs, cachedTags, found := cache.getCachedData("table1")
247 + cachedIndexes, found := cache.getCachedIndexes("table1")
248 require.True(t, found)
249
250 // Cached data should not have the modifications
287 - assert.NotContains(t, cachedOIDs["1"], "col2")
288 - assert.NotContains(t, cachedTags["1"], "tag2")
251 + assert.Equal(t, []string{"1", "2", "3"}, cachedIndexes)
252 + assert.NotContains(t, cachedIndexes, "999")
253 + assert.NotContains(t, cachedIndexes, "4")
254 }
255
256 func TestTableCacheDependencyCleanup(t *testing.T) {
@@ -295,11 +260,11 @@ func TestTableCacheDependencyCleanup(t *testing.T) {
260 table1 := "1.3.6.1.2.1.1"
261 table2 := "1.3.6.1.2.1.2"
262
298 - data := map[string]map[string]string{"1": {"col": "val"}}
263 + indexes := []string{"1"}
264
265 // Cache with circular deps
301 - cache.cacheDataWithDeps(table1, data, nil, []string{table2})
302 - cache.cacheDataWithDeps(table2, data, nil, []string{table1})
266 + cache.cacheIndexesWithDeps(table1, indexes, []string{table2})
267 + cache.cacheIndexesWithDeps(table2, indexes, []string{table1})
268
269 // Check initial state
270 tables, withDeps, totalDeps := cache.stats()
@@ -327,17 +292,17 @@ func TestTableCacheNonExistentDependency(t *testing.T) {
292 cache := newTableCache(100*time.Millisecond, 0)
293
294 // Cache tableA with dependency on non-existent tableB
330 - cache.cacheDataWithDeps("tableA",
331 - map[string]map[string]string{"1": {"col": "val"}},
332 - nil,
295 + cache.cacheIndexesWithDeps("tableA",
296 + []string{"1", "2"},
297 []string{"tableB"})
298
299 // tableA should be cached
336 - _, _, found := cache.getCachedData("tableA")
300 + indexes, found := cache.getCachedIndexes("tableA")
301 assert.True(t, found)
302 + assert.Equal(t, []string{"1", "2"}, indexes)
303
304 // tableB should not be cached
340 - _, _, found = cache.getCachedData("tableB")
305 + _, found = cache.getCachedIndexes("tableB")
306 assert.False(t, found)
307
308 // Dependencies should exist
@@ -345,9 +310,8 @@ func TestTableCacheNonExistentDependency(t *testing.T) {
310 assert.Contains(t, cache.getDependencies("tableB"), "tableA")
311
312 // Now cache tableB
348 - cache.cacheDataWithDeps("tableB",
349 - map[string]map[string]string{"1": {"col": "val"}},
350 - nil,
313 + cache.cacheIndexesWithDeps("tableB",
314 + []string{"1", "2"},
315 []string{"tableA"})
316
317 // Both should be cached
@@ -362,3 +326,60 @@ func TestTableCacheNonExistentDependency(t *testing.T) {
326 assert.Contains(t, expired, "tableA")
327 assert.Contains(t, expired, "tableB")
328 }
329 +
330 +// Helper methods that need to be added to tableCache for tests
331 +func (tc *tableCache) areTablesCached(tableOIDs []string) bool {
332 + tc.mu.RLock()
333 + defer tc.mu.RUnlock()
334 +
335 + if tc.baseTTL == 0 {
336 + return false
337 + }
338 +
339 + now := time.Now()
340 + for _, tableOID := range tableOIDs {
341 + timestamp, ok := tc.timestamps[tableOID]
342 + if !ok {
343 + return false
344 + }
345 +
346 + ttl, ok := tc.tableTTLs[tableOID]
347 + if !ok || now.Sub(timestamp) > ttl {
348 + return false
349 + }
350 + }
351 +
352 + return true
353 +}
354 +
355 +func (tc *tableCache) getDependencies(tableOID string) []string {
356 + tc.mu.RLock()
357 + defer tc.mu.RUnlock()
358 +
359 + deps, ok := tc.tableDeps[tableOID]
360 + if !ok {
361 + return nil
362 + }
363 +
364 + result := make([]string, 0, len(deps))
365 + for dep := range deps {
366 + result = append(result, dep)
367 + }
368 + return result
369 +}
370 +
371 +func (tc *tableCache) stats() (tables int, withDeps int, totalDeps int) {
372 + tc.mu.RLock()
373 + defer tc.mu.RUnlock()
374 +
375 + tables = len(tc.tableIndexes)
376 +
377 + for _, deps := range tc.tableDeps {
378 + if len(deps) > 0 {
379 + withDeps++
380 + totalDeps += len(deps)
381 + }
382 + }
383 +
384 + return tables, withDeps, totalDeps
385 +}
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/utils.go
+9 -5
@@ -252,14 +252,18 @@ func isMappingKeysNumeric(mapping map[string]string) bool {
252 }
253 return true
254 }
255 -func cleanTags(metrics []*ProfileMetrics) {
256 - for _, pm := range metrics {
257 - for _, m := range pm.Metrics {
255 +func cleanMetrics(pms []*ProfileMetrics) {
256 + for _, pm := range pms {
257 + for i := range pm.Metrics {
258 + m := &pm.Metrics[i]
259 + m.Description = metricMetaReplacer.Replace(m.Description)
260 + m.Family = metricMetaReplacer.Replace(m.Family)
261 + m.Unit = metricMetaReplacer.Replace(m.Unit)
262 for k, v := range m.Tags {
259 - m.Tags[k] = tagReplacer.Replace(v)
263 + m.Tags[k] = metricMetaReplacer.Replace(v)
264 }
265 }
266 }
267 }
268
265 -var tagReplacer = strings.NewReplacer("'", "", "\n", " ", "\r", " ")
269 +var metricMetaReplacer = strings.NewReplacer("'", "", "\n", " ", "\r", " ")