@cryptotaxi247 / netdata-1 / commits / 7019e80a5

improvement(go.d/ddsnmp): add table metrics and tags caching optimization (#20465)

Ilya Mashchenko committed Jun 11, 2025 at 12:52 UTC 7019e80a5bac4736dc14e464919b735b2596d3da
9 files changed +512 -133
src/go/plugin/go.d/collector/snmp/charts.go
+2 -11
@@ -315,7 +315,7 @@ func newUserInputChart(cfg ChartConfig) (*module.Chart, error) {
315 return chart, nil
316 }
317
318 -func (c *Collector) addProfileScalarMetricChart(m ddsnmpcollector.Metric) {
318 +func (c *Collector) addProfileScalarMetricChart(pm *ddsnmpcollector.ProfileMetrics, m ddsnmpcollector.Metric) {
319 if m.Name == "" {
320 return
321 }
@@ -343,7 +343,7 @@ func (c *Collector) addProfileScalarMetricChart(m ddsnmpcollector.Metric) {
343 "vendor": c.sysInfo.Organization,
344 "sysName": c.sysInfo.Name,
345 }
346 - maps.Copy(tags, m.Tags)
346 + maps.Copy(tags, pm.Tags)
347 for k, v := range tags {
348 chart.Labels = append(chart.Labels, module.Label{Key: k, Value: v})
349 }
@@ -365,15 +365,6 @@ func (c *Collector) addProfileScalarMetricChart(m ddsnmpcollector.Metric) {
365 }
366 }
367
368 -func (c *Collector) removeProfileScalarMetricChart(metricName string) {
369 - r := strings.NewReplacer(".", "_", " ", "_")
370 - id := fmt.Sprintf("snmp_device_prof_%s", r.Replace(metricName))
371 - if chart := c.Charts().Get(id); chart != nil {
372 - chart.MarkRemove()
373 - chart.MarkNotCreated()
374 - }
375 -}
376 -
368 func dimAlgoFromDdSnmpType(m ddsnmpcollector.Metric) module.DimAlgo {
369 if m.MetricType == ddprofiledefinition.ProfileMetricTypeGauge {
370 return module.Absolute
src/go/plugin/go.d/collector/snmp/collect_profiles.go
+1 -11
@@ -22,18 +22,15 @@ func (c *Collector) collectProfiles(mx map[string]int64) error {
22 return err
23 }
24
25 - seen := make(map[string]bool)
26 -
25 for _, pm := range profMetrics {
26 for _, m := range pm.Metrics {
27 if m.IsTable {
28 continue
29 }
30
33 - seen[m.Name] = true
31 if !c.seenScalarMetrics[m.Name] {
32 c.seenScalarMetrics[m.Name] = true
36 - c.addProfileScalarMetricChart(m)
33 + c.addProfileScalarMetricChart(pm, m)
34 }
35
36 if len(m.Mappings) > 0 {
@@ -48,12 +45,5 @@ func (c *Collector) collectProfiles(mx map[string]int64) error {
45 }
46 }
47
51 - for name := range c.seenScalarMetrics {
52 - if !seen[name] {
53 - delete(c.seenScalarMetrics, name)
54 - c.removeProfileScalarMetricChart(name)
55 - }
56 - }
57 -
48 return nil
49 }
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/collect_scalar.go
+3 -3
@@ -84,18 +84,18 @@ func (c *Collector) collectScalarMetric(cfg ddprofiledefinition.MetricsConfig, p
84 return nil, fmt.Errorf("error processing value for OID %s (%s): %w", cfg.Symbol.Name, cfg.Symbol.OID, err)
85 }
86
87 - tags := make(map[string]string)
87 + staticTags := make(map[string]string)
88
89 for _, tag := range cfg.StaticTags {
90 if n, v, _ := strings.Cut(tag, ":"); n != "" && v != "" {
91 - tags[n] = v
91 + staticTags[n] = v
92 }
93 }
94
95 return &Metric{
96 Name: cfg.Symbol.Name,
97 Value: value,
98 - Tags: tags,
98 + StaticTags: ternary(len(staticTags) > 0, staticTags, nil),
99 Unit: cfg.Symbol.Unit,
100 Description: cfg.Symbol.Description,
101 Family: cfg.Symbol.Family,
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/collect_table.go
+165 -72
@@ -25,6 +25,21 @@ func (c *Collector) collectTableMetrics(prof *ddsnmp.Profile) ([]Metric, error)
25 if cfg.IsScalar() || cfg.Table.OID == "" || doneOids[cfg.Table.OID] {
26 continue
27 }
28 + if func() bool {
29 + for _, tagCfg := range cfg.MetricTags {
30 + if tagCfg.Table != "" && tagCfg.Table != cfg.Table.Name {
31 + c.log.Debugf("Skipping table %s: has cross-table tag from %s", cfg.Table.Name, tagCfg.Table)
32 + return true
33 + }
34 + if len(tagCfg.IndexTransform) > 0 {
35 + c.log.Debugf("Skipping table %s: has index transformation", cfg.Table.Name)
36 + return true
37 + }
38 + }
39 + return false
40 + }() {
41 + continue
42 + }
43 if c.missingOIDs[trimOID(cfg.Table.OID)] {
44 missingOIDs = append(missingOIDs, cfg.Table.OID)
45 continue
@@ -51,15 +66,25 @@ func (c *Collector) collectTableMetrics(prof *ddsnmp.Profile) ([]Metric, error)
66 }
67
68 func (c *Collector) collectSingleTable(cfg ddprofiledefinition.MetricsConfig) ([]Metric, error) {
69 + columnOIDs := make(map[string]ddprofiledefinition.SymbolConfig)
70 + for _, sym := range cfg.Symbols {
71 + columnOIDs[trimOID(sym.OID)] = sym
72 + }
73 +
74 + tagColumnOIDs := make(map[string]ddprofiledefinition.MetricTagConfig)
75 for _, tagCfg := range cfg.MetricTags {
55 - if tagCfg.Table != "" && tagCfg.Table != cfg.Table.Name {
56 - c.log.Debugf("Skipping table %s: has cross-table tag from %s", cfg.Table.Name, tagCfg.Table)
57 - return nil, nil
76 + if tagCfg.Table == "" || tagCfg.Table == cfg.Table.Name {
77 + tagColumnOIDs[trimOID(tagCfg.Symbol.OID)] = tagCfg
78 }
59 - if len(tagCfg.IndexTransform) > 0 {
60 - c.log.Debugf("Skipping table %s: has index transformation", cfg.Table.Name)
61 - return nil, nil
79 + }
80 +
81 + if cachedOIDs, cachedTags, ok := c.tableCache.getCachedData(cfg.Table.OID); ok {
82 + metrics, err := c.collectTableWithCache(cfg, cachedOIDs, cachedTags, columnOIDs)
83 + if err == nil {
84 + c.log.Debugf("Successfully collected table %s using cache", cfg.Table.Name)
85 + return metrics, nil
86 }
87 + c.log.Debugf("Cached collection failed for table %s, falling back to walk: %v", cfg.Table.Name, err)
88 }
89
90 pdus, err := c.snmpWalk(cfg.Table.OID)
@@ -71,28 +96,18 @@ func (c *Collector) collectSingleTable(cfg ddprofiledefinition.MetricsConfig) ([
96 return nil, nil
97 }
98
74 - symColumnOIDs := make(map[string]ddprofiledefinition.SymbolConfig)
75 - for _, sym := range cfg.Symbols {
76 - symColumnOIDs[trimOID(sym.OID)] = sym
77 - }
78 -
79 - tagColumnOIDs := make(map[string]ddprofiledefinition.MetricTagConfig)
80 - for _, tagCfg := range cfg.MetricTags {
81 - if tagCfg.Table == "" || tagCfg.Table == cfg.Table.Name {
82 - tagColumnOIDs[trimOID(tagCfg.Symbol.OID)] = tagCfg
83 - }
84 - }
85 -
86 - allColumnOIDs := make([]string, 0, len(symColumnOIDs)+len(tagColumnOIDs))
87 - for oid := range symColumnOIDs {
99 + allColumnOIDs := make([]string, 0, len(columnOIDs)+len(tagColumnOIDs))
100 + for oid := range columnOIDs {
101 allColumnOIDs = append(allColumnOIDs, oid)
102 }
103 for oid := range tagColumnOIDs {
104 allColumnOIDs = append(allColumnOIDs, oid)
105 }
106
94 - // Group PDUs by row index (index -> column OID -> PDU)
95 - rows := make(map[string]map[string]gosnmp.SnmpPDU, len(pdus)/len(allColumnOIDs))
107 + // Group PDUs by row index and build cache structure
108 + rows := make(map[string]map[string]gosnmp.SnmpPDU)
109 + oidCache := make(map[string]map[string]string) // For caching: index -> column OID -> full OID
110 + tagCache := make(map[string]map[string]string) // For caching: index -> tag name -> value
111
112 for oid, pdu := range pdus {
113 for _, columnOID := range allColumnOIDs {
@@ -101,87 +116,165 @@ func (c *Collector) collectSingleTable(cfg ddprofiledefinition.MetricsConfig) ([
116
117 if rows[index] == nil {
118 rows[index] = make(map[string]gosnmp.SnmpPDU)
119 + oidCache[index] = make(map[string]string)
120 + tagCache[index] = make(map[string]string)
121 }
122 rows[index][columnOID] = pdu
123 + oidCache[index][columnOID] = oid
124 break
125 }
126 }
127 }
128
129 + rowStaticTags := make(map[string]string)
130 + for _, tag := range cfg.StaticTags {
131 + if n, v, _ := strings.Cut(tag, ":"); n != "" && v != "" {
132 + rowStaticTags[n] = v
133 + }
134 + }
135 +
136 var metrics []Metric
137 +
138 for index, rowPDUs := range rows {
113 - rowMetrics, err := c.processTableRow(rowPDUs, symColumnOIDs, tagColumnOIDs, cfg.StaticTags)
114 - if err != nil {
115 - c.log.Debugf("Error processing row %s: %v", index, err)
116 - continue
139 + rowTags := make(map[string]string)
140 +
141 + for columnOID, tagCfg := range tagColumnOIDs {
142 + pdu, ok := rowPDUs[columnOID]
143 + if !ok {
144 + continue
145 + }
146 +
147 + tags, err := processTableMetricTagValue(tagCfg, pdu)
148 + if err != nil {
149 + c.log.Debugf("Error processing tag %s: %v", tagCfg.Tag, err)
150 + continue
151 + }
152 +
153 + for k, v := range tags {
154 + rowTags[k] = v
155 + tagCache[index][k] = v
156 + }
157 + }
158 +
159 + for columnOID, sym := range columnOIDs {
160 + pdu, ok := rowPDUs[columnOID]
161 + if !ok {
162 + continue
163 + }
164 +
165 + value, err := processSymbolValue(sym, pdu)
166 + if err != nil {
167 + c.log.Debugf("Error processing value for %s: %v", sym.Name, err)
168 + continue
169 + }
170 +
171 + metric := Metric{
172 + Name: sym.Name,
173 + Value: value,
174 + StaticTags: ternary(len(rowStaticTags) > 0, rowStaticTags, nil),
175 + Tags: maps.Clone(ternary(len(rowTags) > 0, rowTags, nil)),
176 + Unit: sym.Unit,
177 + Description: sym.Description,
178 + MetricType: getMetricType(sym, pdu),
179 + Family: sym.Family,
180 + Mappings: convSymMappingToNumeric(sym),
181 + IsTable: true,
182 + }
183 +
184 + metrics = append(metrics, metric)
185 }
118 - metrics = append(metrics, rowMetrics...)
186 }
187
188 + c.tableCache.cacheData(cfg.Table.OID, oidCache, tagCache)
189 + c.log.Debugf("Cached table %s structure with %d rows", cfg.Table.Name, len(oidCache))
190 +
191 return metrics, nil
192 }
193
124 -func (c *Collector) processTableRow(
125 - rowPDUs map[string]gosnmp.SnmpPDU,
194 +func (c *Collector) collectTableWithCache(
195 + cfg ddprofiledefinition.MetricsConfig,
196 + cachedOIDs map[string]map[string]string,
197 + cachedTags map[string]map[string]string,
198 columnOIDs map[string]ddprofiledefinition.SymbolConfig,
127 - tagColumnOIDs map[string]ddprofiledefinition.MetricTagConfig,
128 - staticTags []string,
199 ) ([]Metric, error) {
130 - var metrics []Metric
200 + var oidsToGet []string
201 + oidToLocation := make(map[string]struct{ index, column string }) // full OID -> location
202 +
203 + for index, columns := range cachedOIDs {
204 + for columnOID, fullOID := range columns {
205 + // Only GET metric columns, tags are cached
206 + if _, isMetric := columnOIDs[columnOID]; isMetric {
207 + oidsToGet = append(oidsToGet, fullOID)
208 + oidToLocation[trimOID(fullOID)] = struct{ index, column string }{index, columnOID}
209 + }
210 + }
211 + }
212
132 - // Process tags first to ensure all metrics in the row get the same tags
133 - rowTags := make(map[string]string)
213 + if len(oidsToGet) == 0 {
214 + return nil, nil
215 + }
216
135 - for _, tag := range staticTags {
136 - if n, v, _ := strings.Cut(tag, ":"); n != "" && v != "" {
137 - rowTags[n] = v
138 - }
217 + pdus, err := c.snmpGet(oidsToGet)
218 + if err != nil {
219 + return nil, fmt.Errorf("failed to get cached OIDs: %w", err)
220 }
221
141 - for columnOID, tagCfg := range tagColumnOIDs {
142 - pdu, ok := rowPDUs[columnOID]
143 - if !ok {
144 - continue
145 - }
222 + if len(pdus) < len(oidsToGet)/2 { // If we got less than half, probably table structure changed
223 + return nil, fmt.Errorf("table structure may have changed, got %d/%d PDUs", len(pdus), len(oidsToGet))
224 + }
225
147 - tags, err := processTableMetricTagValue(tagCfg, pdu)
148 - if err != nil {
149 - c.log.Debugf("Error processing tag %s: %v", tagCfg.Tag, err)
150 - continue
151 - }
226 + rowStaticTags := make(map[string]string)
227
153 - for k, v := range tags {
154 - rowTags[k] = v
228 + for _, tag := range cfg.StaticTags {
229 + if n, v, _ := strings.Cut(tag, ":"); n != "" && v != "" {
230 + rowStaticTags[n] = v
231 }
232 }
233
158 - for columnOID, sym := range columnOIDs {
159 - pdu, ok := rowPDUs[columnOID]
160 - if !ok {
161 - continue
162 - }
234 + var metrics []Metric
235
164 - value, err := processSymbolValue(sym, pdu)
165 - if err != nil {
166 - c.log.Debugf("Error processing value for %s: %v", sym.Name, err)
167 - continue
168 - }
236 + for index, columns := range cachedOIDs {
237 + rowTags := make(map[string]string)
238
170 - metric := Metric{
171 - Name: sym.Name,
172 - Value: value,
173 - Tags: make(map[string]string),
174 - Unit: sym.Unit,
175 - Description: sym.Description,
176 - MetricType: getMetricType(sym, pdu),
177 - Family: sym.Family,
178 - Mappings: convSymMappingToNumeric(sym),
179 - IsTable: true,
239 + if tags, ok := cachedTags[index]; ok {
240 + for k, v := range tags {
241 + rowTags[k] = v
242 + }
243 }
244
182 - maps.Copy(metric.Tags, rowTags)
245 + for columnOID, fullOID := range columns {
246 + sym, isMetric := columnOIDs[columnOID]
247 + if !isMetric {
248 + continue
249 + }
250
184 - metrics = append(metrics, metric)
251 + pdu, ok := pdus[trimOID(fullOID)]
252 + if !ok {
253 + c.log.Debugf("Missing PDU for cached OID %s", fullOID)
254 + continue
255 + }
256 +
257 + value, err := processSymbolValue(sym, pdu)
258 + if err != nil {
259 + c.log.Debugf("Error processing value for %s: %v", sym.Name, err)
260 + continue
261 + }
262 +
263 + metric := Metric{
264 + Name: sym.Name,
265 + Value: value,
266 + StaticTags: ternary(len(rowStaticTags) > 0, rowStaticTags, nil),
267 + Tags: ternary(len(rowTags) > 0, rowTags, nil),
268 + Unit: sym.Unit,
269 + Description: sym.Description,
270 + MetricType: getMetricType(sym, pdu),
271 + Family: sym.Family,
272 + Mappings: convSymMappingToNumeric(sym),
273 + IsTable: true,
274 + }
275 +
276 + metrics = append(metrics, metric)
277 + }
278 }
279
280 return metrics, nil
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/collector.go
+29 -21
@@ -9,6 +9,7 @@ import (
9 "maps"
10 "slices"
11 "strings"
12 + "time"
13
14 "github.com/gosnmp/gosnmp"
15
@@ -17,23 +18,26 @@ import (
18 "github.com/netdata/netdata/go/plugins/plugin/go.d/collector/snmp/ddsnmp/ddprofiledefinition"
19 )
20
20 -type ProfileMetrics struct {
21 - Source string
22 - DeviceMetadata map[string]string
23 - Metrics []Metric
24 -}
25 -
26 -type Metric struct {
27 - Name string
28 - Description string
29 - Family string
30 - Unit string
31 - MetricType ddprofiledefinition.ProfileMetricType
32 - Tags map[string]string
33 - Mappings map[int64]string
34 - IsTable bool
35 - Value int64
36 -}
21 +type (
22 + ProfileMetrics struct {
23 + Source string
24 + DeviceMetadata map[string]string
25 + Tags map[string]string
26 + Metrics []Metric
27 + }
28 + Metric struct {
29 + Name string
30 + Description string
31 + Family string
32 + Unit string
33 + MetricType ddprofiledefinition.ProfileMetricType
34 + StaticTags map[string]string
35 + Tags map[string]string
36 + Mappings map[int64]string
37 + IsTable bool
38 + Value int64
39 + }
40 +)
41
42 func New(snmpClient gosnmp.Handler, profiles []*ddsnmp.Profile, log *logger.Logger) *Collector {
43 coll := &Collector{
@@ -41,6 +45,7 @@ func New(snmpClient gosnmp.Handler, profiles []*ddsnmp.Profile, log *logger.Logg
45 snmpClient: snmpClient,
46 profiles: make(map[string]*profileState),
47 missingOIDs: make(map[string]bool),
48 + tableCache: newTableCache(5*time.Minute, 0.2), // 5 min TTL with 20% jitter
49 }
50
51 for _, prof := range profiles {
@@ -57,6 +62,7 @@ type (
62 snmpClient gosnmp.Handler
63 profiles map[string]*profileState
64 missingOIDs map[string]bool
65 + tableCache *tableCache
66
67 doTableMetrics bool
68 }
@@ -72,6 +78,10 @@ func (c *Collector) Collect() ([]*ProfileMetrics, error) {
78 var metrics []*ProfileMetrics
79 var errs []error
80
81 + if expired := c.tableCache.clearExpired(); len(expired) > 0 {
82 + c.log.Debugf("Cleared %d expired table cache entries", len(expired))
83 + }
84 +
85 for _, prof := range c.profiles {
86 if ms, err := c.collectProfile(prof); err != nil {
87 errs = append(errs, err)
@@ -88,6 +98,7 @@ func (c *Collector) Collect() ([]*ProfileMetrics, error) {
98 }
99
100 c.updateMetricFamily(metrics)
101 + cleanTags(metrics)
102
103 return metrics, nil
104 }
@@ -125,13 +136,10 @@ func (c *Collector) collectProfile(ps *profileState) (*ProfileMetrics, error) {
136 metrics = append(metrics, tableMetrics...)
137 }
138
128 - for _, m := range metrics {
129 - maps.Copy(m.Tags, ps.globalTags)
130 - }
131 -
139 return &ProfileMetrics{
140 Source: ps.profile.SourceFile,
141 DeviceMetadata: maps.Clone(ps.deviceMetadata),
142 + Tags: maps.Clone(ps.globalTags),
143 Metrics: metrics,
144 }, nil
145 }
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/collector_test.go
+8 -15
@@ -76,7 +76,6 @@ func TestCollector_Collect(t *testing.T) {
76 {
77 Name: "sysUpTime",
78 Value: 123456,
79 - Tags: map[string]string{},
79 MetricType: "gauge",
80 },
81 // sysName will be skipped because it can't be converted to int64
@@ -140,11 +139,11 @@ func TestCollector_Collect(t *testing.T) {
139 expectedResult: []*ProfileMetrics{
140 {
141 DeviceMetadata: nil,
142 + Tags: map[string]string{"device_vendor": "Cisco IOS"},
143 Metrics: []Metric{
144 {
145 Name: "sysUpTime",
146 Value: 123456,
147 - Tags: map[string]string{"device_vendor": "Cisco IOS"},
147 MetricType: "gauge",
148 },
149 },
@@ -220,7 +219,6 @@ func TestCollector_Collect(t *testing.T) {
219 {
220 Name: "sysUpTime",
221 Value: 123456,
223 - Tags: map[string]string{},
222 MetricType: "gauge",
223 },
224 },
@@ -266,7 +264,6 @@ func TestCollector_Collect(t *testing.T) {
264 {
265 Name: "memoryKilobytes",
266 Value: 1024000, // 1024 * 1000
269 - Tags: map[string]string{},
267 MetricType: "gauge",
268 },
269 },
@@ -395,7 +392,6 @@ func TestCollector_Collect(t *testing.T) {
392 {
393 Name: "sysUpTime",
394 Value: 123456,
398 - Tags: map[string]string{},
395 MetricType: "gauge",
396 },
397 },
@@ -441,7 +437,6 @@ func TestCollector_Collect(t *testing.T) {
437 {
438 Name: "temperature",
439 Value: 25,
444 - Tags: map[string]string{},
440 MetricType: "gauge",
441 },
442 },
@@ -508,11 +503,11 @@ func TestCollector_Collect(t *testing.T) {
503 expectedResult: []*ProfileMetrics{
504 {
505 DeviceMetadata: nil,
506 + Tags: map[string]string{"device_type": "router"},
507 Metrics: []Metric{
508 {
509 Name: "sysUpTime",
510 Value: 123456,
515 - Tags: map[string]string{"device_type": "router"},
511 MetricType: "gauge",
512 },
513 },
@@ -582,7 +577,6 @@ func TestCollector_Collect(t *testing.T) {
577 {
578 Name: "clusterHealth",
579 Value: 1,
585 - Tags: map[string]string{},
580 MetricType: "gauge",
581 Mappings: map[int64]string{
582 0: "OK",
@@ -639,7 +633,6 @@ func TestCollector_Collect(t *testing.T) {
633 {
634 Name: "ifOperStatus",
635 Value: 2,
642 - Tags: map[string]string{},
636 MetricType: "gauge",
637 Mappings: map[int64]string{
638 1: "up",
@@ -697,7 +690,6 @@ func TestCollector_Collect(t *testing.T) {
690 {
691 Name: "fanStatus",
692 Value: 2,
700 - Tags: map[string]string{},
693 MetricType: "gauge",
694 Mappings: map[int64]string{
695 1: "normal",
@@ -752,7 +744,6 @@ func TestCollector_Collect(t *testing.T) {
744 {
745 Name: "ifAdminStatus",
746 Value: 0, // mapped from 2 -> 0
755 - Tags: map[string]string{},
747 MetricType: "gauge",
748 Mappings: map[int64]string{
749 1: "1",
@@ -878,7 +869,6 @@ func TestCollector_Collect(t *testing.T) {
869 {
870 Name: "sysUpTime",
871 Value: 123456,
881 - Tags: map[string]string{},
872 MetricType: "gauge",
873 Mappings: nil, // No mappings
874 },
@@ -1215,10 +1205,12 @@ func TestCollector_Collect(t *testing.T) {
1205 {
1206 Name: "myMetric",
1207 Value: 100,
1218 - Tags: map[string]string{
1208 + StaticTags: map[string]string{
1209 "table_type": "performance",
1210 "source": "snmp",
1221 - "interface": "eth0",
1211 + },
1212 + Tags: map[string]string{
1213 + "interface": "eth0",
1214 },
1215 MetricType: ddprofiledefinition.ProfileMetricTypeGauge,
1216 IsTable: true,
@@ -1299,7 +1291,7 @@ func TestCollector_Collect(t *testing.T) {
1291 {
1292 Name: "ifInOctets",
1293 Value: 2000,
1302 - Tags: map[string]string{}, // No interface tag because it's missing
1294 + Tags: nil, // No interface tag because it's missing
1295 MetricType: ddprofiledefinition.ProfileMetricTypeRate,
1296 IsTable: true,
1297 },
@@ -1320,6 +1312,7 @@ func TestCollector_Collect(t *testing.T) {
1312
1313 collector := New(mockHandler, tc.profiles, logger.New())
1314 collector.doTableMetrics = true
1315 + collector.tableCache.setTTL(0, 0)
1316
1317 result, err := collector.Collect()
1318
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/table_cache.go new
+151
@@ -0,0 +1,151 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package ddsnmpcollector
4 +
5 +import (
6 + "math/rand"
7 + "sync"
8 + "time"
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 +// - Per-table TTL with jitter prevents simultaneous refreshes
16 +
17 +type tableCache struct {
18 + // Table OID -> row index -> column OID -> full OID
19 + tables map[string]map[string]map[string]string
20 +
21 + // Table OID -> when cached
22 + timestamps map[string]time.Time
23 +
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 +
30 + baseTTL time.Duration
31 + jitterPct float64
32 + mu sync.RWMutex
33 + rng *rand.Rand
34 +}
35 +
36 +func newTableCache(baseTTL time.Duration, jitterPct float64) *tableCache {
37 + return &tableCache{
38 + tables: make(map[string]map[string]map[string]string),
39 + timestamps: make(map[string]time.Time),
40 + tableTTLs: make(map[string]time.Duration),
41 + tagValues: make(map[string]map[string]map[string]string),
42 + baseTTL: baseTTL,
43 + jitterPct: jitterPct,
44 + rng: rand.New(rand.NewSource(time.Now().UnixNano())),
45 + }
46 +}
47 +
48 +func (tc *tableCache) calculateTableTTL() time.Duration {
49 + base := float64(tc.baseTTL)
50 + jitter := tc.jitterPct
51 +
52 + // Random jitter between -jitterPct and +jitterPct
53 + // Note: This is called from within lock, so don't acquire lock here
54 + randFloat := tc.rng.Float64()
55 + multiplier := 1.0 + (randFloat*2-1)*jitter
56 +
57 + return time.Duration(base * multiplier)
58 +}
59 +
60 +func (tc *tableCache) getCachedData(tableOID string) (oids map[string]map[string]string, tags map[string]map[string]string, found bool) {
61 + tc.mu.RLock()
62 + defer tc.mu.RUnlock()
63 +
64 + if tc.baseTTL == 0 {
65 + return nil, nil, false
66 + }
67 +
68 + timestamp, exists := tc.timestamps[tableOID]
69 + if !exists {
70 + return nil, nil, false
71 + }
72 +
73 + ttl, exists := tc.tableTTLs[tableOID]
74 + if !exists || time.Since(timestamp) > ttl {
75 + return nil, nil, false
76 + }
77 +
78 + oids = tc.tables[tableOID]
79 + tags = tc.tagValues[tableOID]
80 + return oids, tags, true
81 +}
82 +
83 +func (tc *tableCache) cacheData(tableOID string, oidMap map[string]map[string]string, tagValues map[string]map[string]string) {
84 + tc.mu.Lock()
85 + defer tc.mu.Unlock()
86 +
87 + if tc.baseTTL == 0 {
88 + return
89 + }
90 +
91 + // Deep copy the maps to avoid reference issues
92 + oidsCopy := make(map[string]map[string]string, len(oidMap))
93 + for index, columns := range oidMap {
94 + columnsCopy := make(map[string]string, len(columns))
95 + for colOID, fullOID := range columns {
96 + columnsCopy[colOID] = fullOID
97 + }
98 + oidsCopy[index] = columnsCopy
99 + }
100 +
101 + tagsCopy := make(map[string]map[string]string, len(tagValues))
102 + for index, tags := range tagValues {
103 + tagCopy := make(map[string]string, len(tags))
104 + for name, value := range tags {
105 + tagCopy[name] = value
106 + }
107 + tagsCopy[index] = tagCopy
108 + }
109 +
110 + tc.tables[tableOID] = oidsCopy
111 + tc.tagValues[tableOID] = tagsCopy
112 + tc.timestamps[tableOID] = time.Now()
113 + tc.tableTTLs[tableOID] = tc.calculateTableTTL()
114 +}
115 +
116 +func (tc *tableCache) clearExpired() []string {
117 + tc.mu.Lock()
118 + defer tc.mu.Unlock()
119 +
120 + var expired []string
121 + now := time.Now()
122 +
123 + for tableOID, timestamp := range tc.timestamps {
124 + ttl := tc.tableTTLs[tableOID]
125 + if now.Sub(timestamp) > ttl {
126 + delete(tc.tables, tableOID)
127 + delete(tc.timestamps, tableOID)
128 + delete(tc.tableTTLs, tableOID)
129 + delete(tc.tagValues, tableOID)
130 + expired = append(expired, tableOID)
131 + }
132 + }
133 +
134 + return expired
135 +}
136 +
137 +func (tc *tableCache) setTTL(baseTTL time.Duration, jitterPct float64) {
138 + tc.mu.Lock()
139 + defer tc.mu.Unlock()
140 +
141 + tc.baseTTL = baseTTL
142 + tc.jitterPct = jitterPct
143 +
144 + if baseTTL == 0 {
145 + // Clear cache if caching is disabled
146 + tc.tables = make(map[string]map[string]map[string]string)
147 + tc.timestamps = make(map[string]time.Time)
148 + tc.tableTTLs = make(map[string]time.Duration)
149 + tc.tagValues = make(map[string]map[string]map[string]string)
150 + }
151 +}
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/table_cache_test.go new
+142
@@ -0,0 +1,142 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package ddsnmpcollector
4 +
5 +import (
6 + "testing"
7 + "time"
8 +
9 + "github.com/stretchr/testify/assert"
10 + "github.com/stretchr/testify/require"
11 +)
12 +
13 +func TestTableCache(t *testing.T) {
14 + tests := map[string]struct {
15 + name string
16 + }{
17 + "basic cache operations": {},
18 + }
19 +
20 + for name := range tests {
21 + t.Run(name, func(t *testing.T) {
22 + // Create cache with 100ms TTL and 20% jitter
23 + cache := newTableCache(100*time.Millisecond, 0.2)
24 +
25 + // Test data
26 + tableOID := "1.3.6.1.2.1.2.2"
27 + oidMap := map[string]map[string]string{
28 + "1": {
29 + "1.3.6.1.2.1.2.2.1.2": "1.3.6.1.2.1.2.2.1.2.1",
30 + "1.3.6.1.2.1.2.2.1.10": "1.3.6.1.2.1.2.2.1.10.1",
31 + },
32 + "2": {
33 + "1.3.6.1.2.1.2.2.1.2": "1.3.6.1.2.1.2.2.1.2.2",
34 + "1.3.6.1.2.1.2.2.1.10": "1.3.6.1.2.1.2.2.1.10.2",
35 + },
36 + }
37 + tagValues := map[string]map[string]string{
38 + "1": {"interface": "eth0"},
39 + "2": {"interface": "eth1"},
40 + }
41 +
42 + // Cache data
43 + cache.cacheData(tableOID, oidMap, tagValues)
44 +
45 + // Retrieve cached data - should work
46 + cachedOIDs, cachedTags, found := cache.getCachedData(tableOID)
47 + assert.True(t, found)
48 + assert.Equal(t, oidMap, cachedOIDs)
49 + assert.Equal(t, tagValues, cachedTags)
50 +
51 + // Wait for expiration (considering jitter)
52 + time.Sleep(150 * time.Millisecond)
53 +
54 + // Should be expired now
55 + _, _, found = cache.getCachedData(tableOID)
56 + assert.False(t, found)
57 +
58 + // Clean expired entries
59 + expired := cache.clearExpired()
60 + assert.Contains(t, expired, tableOID)
61 +
62 + // Cache should be empty now
63 + assert.Empty(t, cache.tables)
64 + assert.Empty(t, cache.timestamps)
65 + assert.Empty(t, cache.tableTTLs)
66 + assert.Empty(t, cache.tagValues)
67 + })
68 + }
69 +}
70 +
71 +func TestTableCacheJitter(t *testing.T) {
72 + cache := newTableCache(1*time.Second, 0.2) // 1 second with 20% jitter
73 +
74 + // Calculate multiple TTLs to verify jitter
75 + ttls := make([]time.Duration, 10)
76 + for i := range ttls {
77 + ttls[i] = cache.calculateTableTTL()
78 + time.Sleep(1 * time.Millisecond) // Ensure different timestamps
79 + }
80 +
81 + // All TTLs should be between 800ms and 1200ms (±20%)
82 + for _, ttl := range ttls {
83 + assert.GreaterOrEqual(t, ttl, 800*time.Millisecond)
84 + assert.LessOrEqual(t, ttl, 1200*time.Millisecond)
85 + }
86 +
87 + // Verify they're not all the same (jitter is working)
88 + uniqueTTLs := make(map[time.Duration]bool)
89 + for _, ttl := range ttls {
90 + uniqueTTLs[ttl] = true
91 + }
92 + assert.Greater(t, len(uniqueTTLs), 1, "Expected different TTLs due to jitter")
93 +}
94 +
95 +func TestTableCacheDisabled(t *testing.T) {
96 + cache := newTableCache(0, 0) // Disabled cache
97 +
98 + tableOID := "1.3.6.1.2.1.2.2"
99 + oidMap := map[string]map[string]string{
100 + "1": {"1.3.6.1.2.1.2.2.1.2": "1.3.6.1.2.1.2.2.1.2.1"},
101 + }
102 + tagValues := map[string]map[string]string{
103 + "1": {"interface": "eth0"},
104 + }
105 +
106 + // Try to cache data
107 + cache.cacheData(tableOID, oidMap, tagValues)
108 +
109 + // Should not find anything
110 + _, _, found := cache.getCachedData(tableOID)
111 + assert.False(t, found)
112 +
113 + // Cache should remain empty
114 + assert.Empty(t, cache.tables)
115 +}
116 +
117 +func TestTableCacheDeepCopy(t *testing.T) {
118 + cache := newTableCache(1*time.Hour, 0)
119 +
120 + // Original data
121 + oidMap := map[string]map[string]string{
122 + "1": {"col1": "1.2.3.4.1"},
123 + }
124 + tagValues := map[string]map[string]string{
125 + "1": {"tag1": "value1"},
126 + }
127 +
128 + // Cache the data
129 + cache.cacheData("table1", oidMap, tagValues)
130 +
131 + // Modify original maps
132 + oidMap["1"]["col2"] = "should not appear"
133 + tagValues["1"]["tag2"] = "should not appear"
134 +
135 + // Retrieve cached data
136 + cachedOIDs, cachedTags, found := cache.getCachedData("table1")
137 + require.True(t, found)
138 +
139 + // Cached data should not have the modifications
140 + assert.NotContains(t, cachedOIDs["1"], "col2")
141 + assert.NotContains(t, cachedTags["1"], "tag2")
142 +}
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/utils.go
+11
@@ -250,3 +250,14 @@ func isMappingKeysNumeric(mapping map[string]string) bool {
250 }
251 return true
252 }
253 +func cleanTags(metrics []*ProfileMetrics) {
254 + for _, pm := range metrics {
255 + for _, m := range pm.Metrics {
256 + for k, v := range m.Tags {
257 + m.Tags[k] = tagReplacer.Replace(v)
258 + }
259 + }
260 + }
261 +}
262 +
263 +var tagReplacer = strings.NewReplacer("'", "", "\n", " ", "\r", " ")