chore(go.d/ddsnmp): add dependency-based expiration to table cache (#20474)
Ilya Mashchenko committed
Jun 13, 2025 at 12:47 UTC
f8f0581e95583129e05a41b3dc32a71e1abefc4c
2 files changed
+344
-8
src/go/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector/table_cache.go
+121
-7
@@ -12,7 +12,7 @@ import (
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
15
+// - Tables with dependencies expire together to maintain consistency
16
17
type tableCache struct {
18
// Table OID -> row index -> column OID -> full OID
@@ -27,6 +27,10 @@ type tableCache struct {
27
// Table OID -> tag values (index -> tag name -> value)
28
tagValues map[string]map[string]map[string]string
29
30
+ // Table OID -> list of dependent table OIDs (bidirectional)
31
+ // If table A depends on table B, both A->B and B->A are stored
32
+ tableDeps map[string]map[string]bool
33
+
34
baseTTL time.Duration
35
jitterPct float64
36
mu sync.RWMutex
@@ -39,6 +43,7 @@ func newTableCache(baseTTL time.Duration, jitterPct float64) *tableCache {
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())),
@@ -50,7 +55,6 @@ func (tc *tableCache) calculateTableTTL() time.Duration {
55
jitter := tc.jitterPct
56
57
// Random jitter between 0 and +jitterPct
53
- // Note: This is called from within lock, so don't acquire lock here
58
randFloat := tc.rng.Float64()
59
multiplier := 1.0 + randFloat*jitter
60
@@ -81,6 +85,10 @@ func (tc *tableCache) getCachedData(tableOID string) (oids map[string]map[string
85
}
86
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)
89
+}
90
+
91
+func (tc *tableCache) cacheDataWithDeps(tableOID string, oidMap map[string]map[string]string, tagValues map[string]map[string]string, dependencies []string) {
92
tc.mu.Lock()
93
defer tc.mu.Unlock()
94
@@ -111,6 +119,24 @@ func (tc *tableCache) cacheData(tableOID string, oidMap map[string]map[string]st
119
tc.tagValues[tableOID] = tagsCopy
120
tc.timestamps[tableOID] = time.Now()
121
tc.tableTTLs[tableOID] = tc.calculateTableTTL()
122
+
123
+ // Set up bidirectional dependencies
124
+ if len(dependencies) > 0 {
125
+ if tc.tableDeps[tableOID] == nil {
126
+ tc.tableDeps[tableOID] = make(map[string]bool)
127
+ }
128
+
129
+ for _, depTable := range dependencies {
130
+ // Add forward dependency
131
+ tc.tableDeps[tableOID][depTable] = true
132
+
133
+ // Add reverse dependency
134
+ if tc.tableDeps[depTable] == nil {
135
+ tc.tableDeps[depTable] = make(map[string]bool)
136
+ }
137
+ tc.tableDeps[depTable][tableOID] = true
138
+ }
139
+ }
140
}
141
142
func (tc *tableCache) clearExpired() []string {
@@ -120,17 +146,46 @@ func (tc *tableCache) clearExpired() []string {
146
var expired []string
147
now := time.Now()
148
149
+ // First pass: find naturally expired tables
150
+ expiredTables := make(map[string]bool)
151
for tableOID, timestamp := range tc.timestamps {
152
ttl := tc.tableTTLs[tableOID]
153
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)
154
+ expiredTables[tableOID] = true
155
+ }
156
+ }
157
+
158
+ // Second pass: cascade expiration to dependent tables
159
+ for tableOID := range expiredTables {
160
+ for dep := range tc.tableDeps[tableOID] {
161
+ expiredTables[dep] = true
162
}
163
}
164
165
+ // Clear all expired tables
166
+ for tableOID := range expiredTables {
167
+ delete(tc.tables, tableOID)
168
+ delete(tc.timestamps, tableOID)
169
+ delete(tc.tableTTLs, tableOID)
170
+ delete(tc.tagValues, tableOID)
171
+
172
+ // Clean up dependencies
173
+ if deps, ok := tc.tableDeps[tableOID]; ok {
174
+ // Remove this table from other tables' dependency lists
175
+ for depTable := range deps {
176
+ if otherDeps, ok := tc.tableDeps[depTable]; ok {
177
+ delete(otherDeps, tableOID)
178
+ if len(otherDeps) == 0 {
179
+ delete(tc.tableDeps, depTable)
180
+ }
181
+ }
182
+ }
183
+ delete(tc.tableDeps, tableOID)
184
+ }
185
+
186
+ expired = append(expired, tableOID)
187
+ }
188
+
189
return expired
190
}
191
@@ -147,5 +202,64 @@ func (tc *tableCache) setTTL(baseTTL time.Duration, jitterPct float64) {
202
tc.timestamps = make(map[string]time.Time)
203
tc.tableTTLs = make(map[string]time.Duration)
204
tc.tagValues = make(map[string]map[string]map[string]string)
205
+ tc.tableDeps = make(map[string]map[string]bool)
206
+ }
207
+}
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
+223
-1
@@ -19,7 +19,6 @@ func TestTableCache(t *testing.T) {
19
20
for name := range tests {
21
t.Run(name, func(t *testing.T) {
22
- // Create cache with 100ms TTL and 20% jitter
22
cache := newTableCache(100*time.Millisecond, 0.2)
23
24
// Test data
@@ -68,6 +67,146 @@ func TestTableCache(t *testing.T) {
67
}
68
}
69
70
+func TestTableCacheDependencies(t *testing.T) {
71
+ cache := newTableCache(200*time.Millisecond, 0)
72
+
73
+ // Test data for three related tables
74
+ table1OID := "1.3.6.1.2.1.2.2" // ifTable
75
+ table2OID := "1.3.6.1.2.1.31.1.1" // ifXTable
76
+ table3OID := "1.3.6.1.4.1.9.9.276" // cieIfInterfaceTable
77
+
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
+ }
95
+
96
+ // Cache tables with dependencies
97
+ // table1 and table2 depend on each other
98
+ cache.cacheDataWithDeps(table1OID, oidMap1, tagValues1, []string{table2OID})
99
+ cache.cacheDataWithDeps(table2OID, oidMap2, tagValues2, []string{table1OID})
100
+
101
+ // table3 depends on table2
102
+ cache.cacheDataWithDeps(table3OID, oidMap3, nil, []string{table2OID})
103
+
104
+ // All tables should be cached
105
+ assert.True(t, cache.areTablesCached([]string{table1OID, table2OID, table3OID}))
106
+
107
+ // Check dependencies
108
+ deps1 := cache.getDependencies(table1OID)
109
+ assert.Contains(t, deps1, table2OID)
110
+
111
+ deps2 := cache.getDependencies(table2OID)
112
+ assert.Contains(t, deps2, table1OID)
113
+ assert.Contains(t, deps2, table3OID) // Bidirectional
114
+
115
+ deps3 := cache.getDependencies(table3OID)
116
+ assert.Contains(t, deps3, table2OID)
117
+
118
+ // Get cache stats
119
+ tables, withDeps, totalDeps := cache.stats()
120
+ assert.Equal(t, 3, tables)
121
+ assert.Equal(t, 3, withDeps)
122
+ assert.Equal(t, 4, totalDeps) // 1->2, 2->1, 2->3, 3->2
123
+
124
+ // Sleep to let table1 expire naturally
125
+ time.Sleep(220 * time.Millisecond)
126
+
127
+ // Clear expired - should cascade to all dependent tables
128
+ expired := cache.clearExpired()
129
+
130
+ // All three tables should be expired due to dependencies
131
+ assert.Len(t, expired, 3)
132
+ assert.Contains(t, expired, table1OID)
133
+ assert.Contains(t, expired, table2OID)
134
+ assert.Contains(t, expired, table3OID)
135
+
136
+ // Cache should be empty
137
+ assert.Empty(t, cache.tables)
138
+ assert.Empty(t, cache.tableDeps)
139
+}
140
+
141
+func TestTableCacheDependenciesCascade(t *testing.T) {
142
+ cache := newTableCache(100*time.Millisecond, 0)
143
+
144
+ // Create a chain: A -> B -> C -> D
145
+ tableA := "1.3.6.1.2.1.1"
146
+ tableB := "1.3.6.1.2.1.2"
147
+ tableC := "1.3.6.1.2.1.3"
148
+ tableD := "1.3.6.1.2.1.4"
149
+
150
+ data := map[string]map[string]string{"1": {"col": "val"}}
151
+
152
+ // 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})
157
+
158
+ // All should be cached
159
+ assert.True(t, cache.areTablesCached([]string{tableA, tableB, tableC, tableD}))
160
+
161
+ // Wait for A to expire
162
+ time.Sleep(120 * time.Millisecond)
163
+
164
+ // Clear expired - should cascade through entire chain
165
+ expired := cache.clearExpired()
166
+
167
+ // All tables should expire due to cascade
168
+ assert.Len(t, expired, 4)
169
+ assert.Contains(t, expired, tableA)
170
+ assert.Contains(t, expired, tableB)
171
+ assert.Contains(t, expired, tableC)
172
+ assert.Contains(t, expired, tableD)
173
+}
174
+
175
+func TestTableCacheMixedDependencies(t *testing.T) {
176
+ cache := newTableCache(100*time.Millisecond, 0)
177
+
178
+ // Tables with deps
179
+ table1 := "1.3.6.1.2.1.1"
180
+ table2 := "1.3.6.1.2.1.2"
181
+
182
+ // Table without deps
183
+ table3 := "1.3.6.1.2.1.3"
184
+
185
+ data := map[string]map[string]string{"1": {"col": "val"}}
186
+
187
+ // 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
191
+
192
+ // All should be cached
193
+ _, _, found1 := cache.getCachedData(table1)
194
+ _, _, found2 := cache.getCachedData(table2)
195
+ _, _, found3 := cache.getCachedData(table3)
196
+ assert.True(t, found1)
197
+ assert.True(t, found2)
198
+ assert.True(t, found3)
199
+
200
+ // Wait for expiration
201
+ time.Sleep(120 * time.Millisecond)
202
+
203
+ // Clear expired
204
+ expired := cache.clearExpired()
205
+
206
+ // All should be expired (table3 independently, table1&2 together)
207
+ assert.Len(t, expired, 3)
208
+}
209
+
210
func TestTableCacheJitter(t *testing.T) {
211
cache := newTableCache(1*time.Second, 0.2) // 1 second with 20% jitter
212
@@ -110,6 +249,14 @@ func TestTableCacheDisabled(t *testing.T) {
249
_, _, found := cache.getCachedData(tableOID)
250
assert.False(t, found)
251
252
+ // Try to cache with dependencies
253
+ cache.cacheDataWithDeps(tableOID, oidMap, tagValues, []string{"other.table"})
254
+
255
+ // Should not find anything
256
+ _, _, found = cache.getCachedData(tableOID)
257
+ assert.False(t, found)
258
+ assert.False(t, cache.areTablesCached([]string{tableOID}))
259
+
260
// Cache should remain empty
261
assert.Empty(t, cache.tables)
262
}
@@ -140,3 +287,78 @@ func TestTableCacheDeepCopy(t *testing.T) {
287
assert.NotContains(t, cachedOIDs["1"], "col2")
288
assert.NotContains(t, cachedTags["1"], "tag2")
289
}
290
+
291
+func TestTableCacheDependencyCleanup(t *testing.T) {
292
+ cache := newTableCache(100*time.Millisecond, 0)
293
+
294
+ // Create circular dependencies
295
+ table1 := "1.3.6.1.2.1.1"
296
+ table2 := "1.3.6.1.2.1.2"
297
+
298
+ data := map[string]map[string]string{"1": {"col": "val"}}
299
+
300
+ // Cache with circular deps
301
+ cache.cacheDataWithDeps(table1, data, nil, []string{table2})
302
+ cache.cacheDataWithDeps(table2, data, nil, []string{table1})
303
+
304
+ // Check initial state
305
+ tables, withDeps, totalDeps := cache.stats()
306
+ assert.Equal(t, 2, tables)
307
+ assert.Equal(t, 2, withDeps)
308
+ assert.Equal(t, 2, totalDeps)
309
+
310
+ // Wait for expiration
311
+ time.Sleep(120 * time.Millisecond)
312
+
313
+ // Clear expired
314
+ cache.clearExpired()
315
+
316
+ // Check cleanup
317
+ tables, withDeps, totalDeps = cache.stats()
318
+ assert.Equal(t, 0, tables)
319
+ assert.Equal(t, 0, withDeps)
320
+ assert.Equal(t, 0, totalDeps)
321
+
322
+ // Dependencies should be cleaned up
323
+ assert.Empty(t, cache.tableDeps)
324
+}
325
+
326
+func TestTableCacheNonExistentDependency(t *testing.T) {
327
+ cache := newTableCache(100*time.Millisecond, 0)
328
+
329
+ // Cache tableA with dependency on non-existent tableB
330
+ cache.cacheDataWithDeps("tableA",
331
+ map[string]map[string]string{"1": {"col": "val"}},
332
+ nil,
333
+ []string{"tableB"})
334
+
335
+ // tableA should be cached
336
+ _, _, found := cache.getCachedData("tableA")
337
+ assert.True(t, found)
338
+
339
+ // tableB should not be cached
340
+ _, _, found = cache.getCachedData("tableB")
341
+ assert.False(t, found)
342
+
343
+ // Dependencies should exist
344
+ assert.Contains(t, cache.getDependencies("tableA"), "tableB")
345
+ assert.Contains(t, cache.getDependencies("tableB"), "tableA")
346
+
347
+ // Now cache tableB
348
+ cache.cacheDataWithDeps("tableB",
349
+ map[string]map[string]string{"1": {"col": "val"}},
350
+ nil,
351
+ []string{"tableA"})
352
+
353
+ // Both should be cached
354
+ assert.True(t, cache.areTablesCached([]string{"tableA", "tableB"}))
355
+
356
+ // Wait for expiration
357
+ time.Sleep(120 * time.Millisecond)
358
+
359
+ // Clear expired - both should expire together
360
+ expired := cache.clearExpired()
361
+ assert.Len(t, expired, 2)
362
+ assert.Contains(t, expired, "tableA")
363
+ assert.Contains(t, expired, "tableB")
364
+}