johanels/akvorado
1
0
Fork 0
mirror of https://github.com/akvorado/akvorado.git synced 2025-12-11 22:14:02 +01:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

inlet/snmp: use netip.Addr internally instead of string

Browse Source 
Also, make SubnetMap use `netip.Addr` as well.
This commit is contained in:
Vincent Bernat
2022-08-27 08:13:28 +02:00
parent 7e3fc03c11
commit 6bfd3a0bd0
12 changed files with 147 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
common/helpers/subnetmap.go
View File

@@ -6,6 +6,7 @@ package helpers
import (
"fmt"
"net"
"net/netip"
"reflect"
"regexp"
"strings"
@@ -25,19 +26,18 @@ type SubnetMap[V any] struct {
// Lookup will search for the most specific subnet matching the
// provided IP address and return the value associated with it.
func (sm *SubnetMap[V]) Lookup(ip net.IP) (V, bool) {
func (sm *SubnetMap[V]) Lookup(ip netip.Addr) (V, bool) {
if sm == nil || sm.tree == nil {
var value V
return value, false
}
ip = ip.To16()
ok, value := sm.tree.FindDeepestTag(patricia.NewIPv6Address(ip, 128))
ok, value := sm.tree.FindDeepestTag(patricia.NewIPv6Address(ip.AsSlice(), 128))
return value, ok
}
// LookupOrDefault calls lookup and if not found, will return the
// provided default value.
func (sm *SubnetMap[V]) LookupOrDefault(ip net.IP, fallback V) V {
func (sm *SubnetMap[V]) LookupOrDefault(ip netip.Addr, fallback V) V {
if value, ok := sm.Lookup(ip); ok {
return value
}

33
common/helpers/subnetmap_test.go
View File

@@ -4,7 +4,7 @@
package helpers_test
import (
"net"
"net/netip"
"testing"
"github.com/gin-gonic/gin"
@@ -27,22 +27,23 @@ func TestSubnetMapUnmarshalHook(t *testing.T) {
Description: "nil",
Input: nilMap,
Tests: map[string]string{
"203.0.113.1": "",
"::ffff:203.0.113.1": "",
},
}, {
Description: "empty",
Input: gin.H{},
Tests: map[string]string{
"203.0.113.1": "",
"::ffff:203.0.113.1": "",
},
}, {
Description: "IPv4 subnet",
Input: gin.H{"203.0.113.0/24": "customer1"},
Tests: map[string]string{
"::ffff:203.0.113.18": "customer1",
"203.0.113.16": "customer1",
"203.0.1.1": "",
"::ffff:203.0.113.16": "customer1",
"203.0.113.16": "",
"::ffff:203.0.1.1": "",
"203.0.1.1": "",
"2001:db8:1::12": "",
},
}, {
@@ -50,7 +51,6 @@ func TestSubnetMapUnmarshalHook(t *testing.T) {
Input: gin.H{"203.0.113.1": "customer1"},
Tests: map[string]string{
"::ffff:203.0.113.1": "customer1",
"203.0.113.1": "customer1",
"2001:db8:1::12": "",
},
YAML: gin.H{"203.0.113.1/32": "customer1"},
@@ -67,9 +67,7 @@ func TestSubnetMapUnmarshalHook(t *testing.T) {
Input: gin.H{"::ffff:203.0.113.0/120": "customer2"},
Tests: map[string]string{
"::ffff:203.0.113.10": "customer2",
"203.0.113.10": "customer2",
"::ffff:203.0.112.10": "",
"203.0.112.10": "",
},
YAML: gin.H{"203.0.113.0/24": "customer2"},
}, {
@@ -105,7 +103,7 @@ func TestSubnetMapUnmarshalHook(t *testing.T) {
Description: "Single value",
Input: "customer",
Tests: map[string]string{
"203.0.113.4": "customer",
"::ffff:203.0.113.4": "customer",
"2001:db8::1": "customer",
},
YAML: map[string]string{
@@ -140,7 +138,7 @@ func TestSubnetMapUnmarshalHook(t *testing.T) {
}
got := map[string]string{}
for k := range tc.Tests {
v, _ := tree.Lookup(net.ParseIP(k))
v, _ := tree.Lookup(netip.MustParseAddr(k))
got[k] = v
}
if diff := helpers.Diff(got, tc.Tests); diff != "" {
@@ -231,3 +229,18 @@ func TestSubnetMapUnmarshalHookWithMapValue(t *testing.T) {
})
}
}
func TestToMap(t *testing.T) {
input := helpers.MustNewSubnetMap(map[string]string{
"2001:db8::/64": "hello",
"::ffff:192.0.2.0/120": "bye",
})
got := input.ToMap()
expected := map[string]string{
"2001:db8::/64": "hello",
"192.0.2.0/24": "bye",
}
if diff := helpers.Diff(got, expected); diff != "" {
t.Fatalf("ToMap() (-got, +want):\n%s", diff)
}
}

2
common/helpers/tests.go
View File

@@ -16,6 +16,7 @@ import (
"fmt"
"net"
"net/http"
"net/netip"
"os"
"reflect"
"testing"
@@ -34,6 +35,7 @@ var prettyC = pretty.Config{
IncludeUnexported: false,
Formatter: map[reflect.Type]interface{}{
reflect.TypeOf(net.IP{}): fmt.Sprint,
reflect.TypeOf(netip.Addr{}): fmt.Sprint,
reflect.TypeOf(time.Time{}): fmt.Sprint,
reflect.TypeOf(SubnetMap[string]{}): fmt.Sprint,
},

7
inlet/core/hydrate.go
View File

@@ -6,6 +6,7 @@ package core
import (
"fmt"
"net"
"net/netip"
"strconv"
"time"
@@ -16,11 +17,11 @@ import (
)
// hydrateFlow adds more data to a flow.
func (c *Component) hydrateFlow(exporterIP net.IP, exporterStr string, flow *flow.Message) (skip bool) {
func (c *Component) hydrateFlow(exporterIP netip.Addr, exporterStr string, flow *flow.Message) (skip bool) {
errLogger := c.r.Sample(reporter.BurstSampler(time.Minute, 10))
if flow.InIf != 0 {
exporterName, iface, err := c.d.Snmp.Lookup(exporterStr, uint(flow.InIf))
exporterName, iface, err := c.d.Snmp.Lookup(exporterIP, uint(flow.InIf))
if err != nil {
if err != snmp.ErrCacheMiss {
errLogger.Err(err).Str("exporter", exporterStr).Msg("unable to query SNMP cache")
@@ -36,7 +37,7 @@ func (c *Component) hydrateFlow(exporterIP net.IP, exporterStr string, flow *flo
}
if flow.OutIf != 0 {
exporterName, iface, err := c.d.Snmp.Lookup(exporterStr, uint(flow.OutIf))
exporterName, iface, err := c.d.Snmp.Lookup(exporterIP, uint(flow.OutIf))
if err != nil {
// Only register a cache miss if we don't have one.
// TODO: maybe we could do one SNMP query for both interfaces.

4
inlet/core/root.go
View File

@@ -7,6 +7,7 @@ package core
import (
"fmt"
"net"
"net/netip"
"sync/atomic"
"time"
@@ -119,7 +120,8 @@ func (c *Component) runWorker(workerID int) error {
c.metrics.flowsReceived.WithLabelValues(exporter).Inc()
// Hydratation
if skip := c.hydrateFlow(flow.ExporterAddress, exporter, flow); skip {
ip, _ := netip.AddrFromSlice(flow.ExporterAddress)
if skip := c.hydrateFlow(ip, exporter, flow); skip {
continue
}

23
inlet/snmp/cache.go
View File

@@ -9,6 +9,7 @@ import (
"errors"
"fmt"
"io/ioutil"
"net/netip"
"os"
"path/filepath"
"sync"
@@ -26,13 +27,13 @@ var (
// ErrCacheVersion is triggered when loading a cache from an incompatible version
ErrCacheVersion = errors.New("SNMP cache version mismatch")
// cacheCurrentVersionNumber is the current version of the on-disk cache format
cacheCurrentVersionNumber = 8
cacheCurrentVersionNumber = 9
)
// snmpCache represents the SNMP cache.
type snmpCache struct {
r *reporter.Reporter
cache map[string]*cachedExporter
cache map[netip.Addr]*cachedExporter
cacheLock sync.RWMutex
clock clock.Clock
@@ -69,7 +70,7 @@ type cachedInterface struct {
func newSNMPCache(r *reporter.Reporter, clock clock.Clock) *snmpCache {
sc := &snmpCache{
r: r,
cache: make(map[string]*cachedExporter),
cache: make(map[netip.Addr]*cachedExporter),
clock: clock,
}
sc.metrics.cacheHit = r.Counter(
@@ -113,11 +114,11 @@ func newSNMPCache(r *reporter.Reporter, clock clock.Clock) *snmpCache {
// Lookup will perform a lookup of the cache. It returns the exporter
// name as well as the requested interface.
func (sc *snmpCache) Lookup(ip string, ifIndex uint) (string, Interface, error) {
func (sc *snmpCache) Lookup(ip netip.Addr, ifIndex uint) (string, Interface, error) {
return sc.lookup(ip, ifIndex, true)
}
func (sc *snmpCache) lookup(ip string, ifIndex uint, touchAccess bool) (string, Interface, error) {
func (sc *snmpCache) lookup(ip netip.Addr, ifIndex uint, touchAccess bool) (string, Interface, error) {
sc.cacheLock.RLock()
defer sc.cacheLock.RUnlock()
exporter, ok := sc.cache[ip]
@@ -138,7 +139,7 @@ func (sc *snmpCache) lookup(ip string, ifIndex uint, touchAccess bool) (string,
}
// Put a new entry in the cache.
func (sc *snmpCache) Put(ip string, exporterName string, ifIndex uint, iface Interface) {
func (sc *snmpCache) Put(ip netip.Addr, exporterName string, ifIndex uint, iface Interface) {
sc.cacheLock.Lock()
defer sc.cacheLock.Unlock()
@@ -181,9 +182,9 @@ func (sc *snmpCache) Expire(older time.Duration) (count uint) {
// Return entries older than the provided duration. If LastAccessed is
// true, rely on last access, otherwise on last update.
func (sc *snmpCache) entriesOlderThan(older time.Duration, lastAccessed bool) map[string]map[uint]Interface {
func (sc *snmpCache) entriesOlderThan(older time.Duration, lastAccessed bool) map[netip.Addr]map[uint]Interface {
threshold := sc.clock.Now().Add(-older).Unix()
result := make(map[string]map[uint]Interface)
result := make(map[netip.Addr]map[uint]Interface)
sc.cacheLock.RLock()
defer sc.cacheLock.RUnlock()
@@ -209,13 +210,13 @@ func (sc *snmpCache) entriesOlderThan(older time.Duration, lastAccessed bool) ma
// Need updates returns a map of interface entries that would need to
// be updated. It relies on last update.
func (sc *snmpCache) NeedUpdates(older time.Duration) map[string]map[uint]Interface {
func (sc *snmpCache) NeedUpdates(older time.Duration) map[netip.Addr]map[uint]Interface {
return sc.entriesOlderThan(older, false)
}
// Need updates returns a map of interface entries that would have
// expired. It relies on last access.
func (sc *snmpCache) WouldExpire(older time.Duration) map[string]map[uint]Interface {
func (sc *snmpCache) WouldExpire(older time.Duration) map[netip.Addr]map[uint]Interface {
return sc.entriesOlderThan(older, true)
}
@@ -284,7 +285,7 @@ func (sc *snmpCache) GobDecode(data []byte) error {
if version != cacheCurrentVersionNumber {
return ErrCacheVersion
}
cache := map[string]*cachedExporter{}
cache := map[netip.Addr]*cachedExporter{}
if err := decoder.Decode(&cache); err != nil {
return err
}

71
inlet/snmp/cache_test.go
View File

@@ -8,6 +8,7 @@ import (
"fmt"
"io/fs"
"math/rand"
"net/netip"
"path/filepath"
"strconv"
"sync"
@@ -37,7 +38,9 @@ type answer struct {
func expectCacheLookup(t *testing.T, sc *snmpCache, exporterIP string, ifIndex uint, expected answer) {
t.Helper()
gotExporterName, gotInterface, err := sc.lookup(exporterIP, ifIndex, false)
ip := netip.MustParseAddr(exporterIP)
ip = netip.AddrFrom16(ip.As16())
gotExporterName, gotInterface, err := sc.lookup(ip, ifIndex, false)
got := answer{gotExporterName, gotInterface, err}
if diff := helpers.Diff(got, expected); diff != "" {
t.Errorf("Lookup() (-got, +want):\n%s", diff)
@@ -63,7 +66,7 @@ func TestGetEmpty(t *testing.T) {
func TestSimpleLookup(t *testing.T) {
r, _, sc := setupTestCache(t)
sc.Put("127.0.0.1", "localhost", 676, Interface{Name: "Gi0/0/0/1", Description: "Transit", Speed: 1000})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.1"), "localhost", 676, Interface{Name: "Gi0/0/0/1", Description: "Transit", Speed: 1000})
expectCacheLookup(t, sc, "127.0.0.1", 676, answer{
ExporterName: "localhost",
Interface: Interface{Name: "Gi0/0/0/1", Description: "Transit", Speed: 1000}})
@@ -85,11 +88,11 @@ func TestSimpleLookup(t *testing.T) {
func TestExpire(t *testing.T) {
r, clock, sc := setupTestCache(t)
sc.Put("127.0.0.1", "localhost", 676, Interface{Name: "Gi0/0/0/1", Description: "Transit"})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.1"), "localhost", 676, Interface{Name: "Gi0/0/0/1", Description: "Transit"})
clock.Add(10 * time.Minute)
sc.Put("127.0.0.1", "localhost2", 678, Interface{Name: "Gi0/0/0/2", Description: "Peering"})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.1"), "localhost2", 678, Interface{Name: "Gi0/0/0/2", Description: "Peering"})
clock.Add(10 * time.Minute)
sc.Put("127.0.0.2", "localhost3", 678, Interface{Name: "Gi0/0/0/1", Description: "IX"})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.2"), "localhost3", 678, Interface{Name: "Gi0/0/0/1", Description: "IX"})
clock.Add(10 * time.Minute)
sc.Expire(time.Hour)
expectCacheLookup(t, sc, "127.0.0.1", 676, answer{
@@ -119,7 +122,7 @@ func TestExpire(t *testing.T) {
expectCacheLookup(t, sc, "127.0.0.1", 676, answer{Err: ErrCacheMiss})
expectCacheLookup(t, sc, "127.0.0.1", 678, answer{Err: ErrCacheMiss})
expectCacheLookup(t, sc, "127.0.0.2", 678, answer{Err: ErrCacheMiss})
sc.Put("127.0.0.1", "localhost", 676, Interface{Name: "Gi0/0/0/1", Description: "Transit"})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.1"), "localhost", 676, Interface{Name: "Gi0/0/0/1", Description: "Transit"})
clock.Add(10 * time.Minute)
sc.Expire(19 * time.Minute)
expectCacheLookup(t, sc, "127.0.0.1", 676, answer{
@@ -141,15 +144,15 @@ func TestExpire(t *testing.T) {
func TestExpireRefresh(t *testing.T) {
_, clock, sc := setupTestCache(t)
sc.Put("127.0.0.1", "localhost", 676, Interface{Name: "Gi0/0/0/1", Description: "Transit"})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.1"), "localhost", 676, Interface{Name: "Gi0/0/0/1", Description: "Transit"})
clock.Add(10 * time.Minute)
sc.Put("127.0.0.1", "localhost", 678, Interface{Name: "Gi0/0/0/2", Description: "Peering"})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.1"), "localhost", 678, Interface{Name: "Gi0/0/0/2", Description: "Peering"})
clock.Add(10 * time.Minute)
sc.Put("127.0.0.2", "localhost2", 678, Interface{Name: "Gi0/0/0/1", Description: "IX"})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.2"), "localhost2", 678, Interface{Name: "Gi0/0/0/1", Description: "IX"})
clock.Add(10 * time.Minute)
// Refresh first entry
sc.Lookup("127.0.0.1", 676)
sc.Lookup(netip.MustParseAddr("::ffff:127.0.0.1"), 676)
clock.Add(10 * time.Minute)
sc.Expire(29 * time.Minute)
@@ -164,14 +167,14 @@ func TestExpireRefresh(t *testing.T) {
func TestWouldExpire(t *testing.T) {
_, clock, sc := setupTestCache(t)
sc.Put("127.0.0.1", "localhost", 676, Interface{Name: "Gi0/0/0/1", Description: "Transit"})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.1"), "localhost", 676, Interface{Name: "Gi0/0/0/1", Description: "Transit"})
clock.Add(10 * time.Minute)
sc.Put("127.0.0.1", "localhost", 678, Interface{Name: "Gi0/0/0/2", Description: "Peering"})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.1"), "localhost", 678, Interface{Name: "Gi0/0/0/2", Description: "Peering"})
clock.Add(10 * time.Minute)
sc.Put("127.0.0.2", "localhost2", 678, Interface{Name: "Gi0/0/0/1", Description: "IX"})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.2"), "localhost2", 678, Interface{Name: "Gi0/0/0/1", Description: "IX"})
clock.Add(10 * time.Minute)
// Refresh
sc.Lookup("127.0.0.1", 676)
sc.Lookup(netip.MustParseAddr("::ffff:127.0.0.1"), 676)
clock.Add(10 * time.Minute)
cases := []struct {
@@ -179,24 +182,24 @@ func TestWouldExpire(t *testing.T) {
Expected map[string]map[uint]Interface
}{
{9, map[string]map[uint]Interface{
"127.0.0.1": {
"::ffff:127.0.0.1": {
676: Interface{Name: "Gi0/0/0/1", Description: "Transit"},
678: Interface{Name: "Gi0/0/0/2", Description: "Peering"},
},
"127.0.0.2": {
"::ffff:127.0.0.2": {
678: Interface{Name: "Gi0/0/0/1", Description: "IX"},
},
}},
{19, map[string]map[uint]Interface{
"127.0.0.1": {
"::ffff:127.0.0.1": {
678: Interface{Name: "Gi0/0/0/2", Description: "Peering"},
},
"127.0.0.2": {
"::ffff:127.0.0.2": {
678: Interface{Name: "Gi0/0/0/1", Description: "IX"},
},
}},
{29, map[string]map[uint]Interface{
"127.0.0.1": {
"::ffff:127.0.0.1": {
678: Interface{Name: "Gi0/0/0/2", Description: "Peering"},
},
}},
@@ -214,14 +217,14 @@ func TestWouldExpire(t *testing.T) {
func TestNeedUpdates(t *testing.T) {
_, clock, sc := setupTestCache(t)
sc.Put("127.0.0.1", "localhost", 676, Interface{Name: "Gi0/0/0/1", Description: "Transit"})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.1"), "localhost", 676, Interface{Name: "Gi0/0/0/1", Description: "Transit"})
clock.Add(10 * time.Minute)
sc.Put("127.0.0.1", "localhost", 678, Interface{Name: "Gi0/0/0/2", Description: "Peering"})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.1"), "localhost", 678, Interface{Name: "Gi0/0/0/2", Description: "Peering"})
clock.Add(10 * time.Minute)
sc.Put("127.0.0.2", "localhost2", 678, Interface{Name: "Gi0/0/0/1", Description: "IX"})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.2"), "localhost2", 678, Interface{Name: "Gi0/0/0/1", Description: "IX"})
clock.Add(10 * time.Minute)
// Refresh
sc.Put("127.0.0.1", "localhost1", 676, Interface{Name: "Gi0/0/0/1", Description: "Transit"})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.1"), "localhost1", 676, Interface{Name: "Gi0/0/0/1", Description: "Transit"})
clock.Add(10 * time.Minute)
cases := []struct {
@@ -229,24 +232,24 @@ func TestNeedUpdates(t *testing.T) {
Expected map[string]map[uint]Interface
}{
{9, map[string]map[uint]Interface{
"127.0.0.1": {
"::ffff:127.0.0.1": {
676: Interface{Name: "Gi0/0/0/1", Description: "Transit"},
678: Interface{Name: "Gi0/0/0/2", Description: "Peering"},
},
"127.0.0.2": {
"::ffff:127.0.0.2": {
678: Interface{Name: "Gi0/0/0/1", Description: "IX"},
},
}},
{19, map[string]map[uint]Interface{
"127.0.0.1": {
"::ffff:127.0.0.1": {
678: Interface{Name: "Gi0/0/0/2", Description: "Peering"},
},
"127.0.0.2": {
"::ffff:127.0.0.2": {
678: Interface{Name: "Gi0/0/0/1", Description: "IX"},
},
}},
{29, map[string]map[uint]Interface{
"127.0.0.1": {
"::ffff:127.0.0.1": {
678: Interface{Name: "Gi0/0/0/2", Description: "Peering"},
},
}},
@@ -272,11 +275,11 @@ func TestLoadNotExist(t *testing.T) {
func TestSaveLoad(t *testing.T) {
_, clock, sc := setupTestCache(t)
sc.Put("127.0.0.1", "localhost", 676, Interface{Name: "Gi0/0/0/1", Description: "Transit"})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.1"), "localhost", 676, Interface{Name: "Gi0/0/0/1", Description: "Transit"})
clock.Add(10 * time.Minute)
sc.Put("127.0.0.1", "localhost", 678, Interface{Name: "Gi0/0/0/2", Description: "Peering"})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.1"), "localhost", 678, Interface{Name: "Gi0/0/0/2", Description: "Peering"})
clock.Add(10 * time.Minute)
sc.Put("127.0.0.2", "localhost2", 678, Interface{Name: "Gi0/0/0/1", Description: "IX", Speed: 1000})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.2"), "localhost2", 678, Interface{Name: "Gi0/0/0/1", Description: "IX", Speed: 1000})
target := filepath.Join(t.TempDir(), "cache")
if err := sc.Save(target); err != nil {
@@ -301,7 +304,7 @@ func TestSaveLoad(t *testing.T) {
func TestLoadMismatchVersion(t *testing.T) {
_, _, sc := setupTestCache(t)
sc.Put("127.0.0.1", "localhost", 676, Interface{Name: "Gi0/0/0/1", Description: "Transit"})
sc.Put(netip.MustParseAddr("::ffff:127.0.0.1"), "localhost", 676, Interface{Name: "Gi0/0/0/1", Description: "Transit"})
target := filepath.Join(t.TempDir(), "cache")
cacheCurrentVersionNumber++
@@ -343,7 +346,7 @@ func TestConcurrentOperations(t *testing.T) {
for {
ip := rand.Intn(10)
iface := rand.Intn(100)
sc.Put(fmt.Sprintf("127.0.0.%d", ip),
sc.Put(netip.MustParseAddr(fmt.Sprintf("::ffff:127.0.0.%d", ip)),
fmt.Sprintf("localhost%d", ip),
uint(iface), Interface{Name: "Gi0/0/0/1", Description: "Transit"})
select {
@@ -362,7 +365,7 @@ func TestConcurrentOperations(t *testing.T) {
for {
ip := rand.Intn(10)
iface := rand.Intn(100)
sc.Lookup(fmt.Sprintf("127.0.0.%d", ip),
sc.Lookup(netip.MustParseAddr(fmt.Sprintf("::ffff:127.0.0.%d", ip)),
uint(iface))
atomic.AddInt64(&lookups, 1)
select {

46
inlet/snmp/poller.go
View File

@@ -7,7 +7,7 @@ import (
"context"
"errors"
"fmt"
"net"
"net/netip"
"sync"
"time"
@@ -19,7 +19,7 @@ import (
)
type poller interface {
Poll(ctx context.Context, exporterIP string, port uint16, ifIndexes []uint) error
Poll(ctx context.Context, exporterIP netip.Addr, port uint16, ifIndexes []uint) error
}
// realPoller will poll exporters using real SNMP requests.
@@ -31,7 +31,7 @@ type realPoller struct {
pendingRequests map[string]struct{}
pendingRequestsLock sync.Mutex
errLogger reporter.Logger
put func(exporterIP, exporterName string, ifIndex uint, iface Interface)
put func(exporterIP netip.Addr, exporterName string, ifIndex uint, iface Interface)
metrics struct {
pendingRequests reporter.GaugeFunc
@@ -50,7 +50,7 @@ type pollerConfig struct {
}
// newPoller creates a new SNMP poller.
func newPoller(r *reporter.Reporter, config pollerConfig, clock clock.Clock, put func(string, string, uint, Interface)) *realPoller {
func newPoller(r *reporter.Reporter, config pollerConfig, clock clock.Clock, put func(netip.Addr, string, uint, Interface)) *realPoller {
p := &realPoller{
r: r,
config: config,
@@ -92,13 +92,14 @@ func newPoller(r *reporter.Reporter, config pollerConfig, clock clock.Clock, put
return p
}
func (p *realPoller) Poll(ctx context.Context, exporter string, port uint16, ifIndexes []uint) error {
func (p *realPoller) Poll(ctx context.Context, exporter netip.Addr, port uint16, ifIndexes []uint) error {
// Check if already have a request running
exporterStr := exporter.Unmap().String()
filteredIfIndexes := make([]uint, 0, len(ifIndexes))
keys := make([]string, 0, len(ifIndexes))
p.pendingRequestsLock.Lock()
for _, ifIndex := range ifIndexes {
key := fmt.Sprintf("%s@%d", exporter, ifIndex)
key := fmt.Sprintf("%s@%d", exporterStr, ifIndex)
_, ok := p.pendingRequests[key]
if !ok {
p.pendingRequests[key] = struct{}{}
@@ -120,20 +121,19 @@ func (p *realPoller) Poll(ctx context.Context, exporter string, port uint16, ifI
}()
// Instantiate an SNMP state
exporterIP := net.ParseIP(exporter)
g := &gosnmp.GoSNMP{
Context: ctx,
Target: exporter,
Target: exporterStr,
Port: port,
Retries: p.config.Retries,
Timeout: p.config.Timeout,
UseUnconnectedUDPSocket: true,
Logger: gosnmp.NewLogger(&goSNMPLogger{p.r}),
OnRetry: func(*gosnmp.GoSNMP) {
p.metrics.retries.WithLabelValues(exporter).Inc()
p.metrics.retries.WithLabelValues(exporterStr).Inc()
},
}
if securityParameters, ok := p.config.SecurityParameters.Lookup(exporterIP); ok {
if securityParameters, ok := p.config.SecurityParameters.Lookup(exporter); ok {
g.Version = gosnmp.Version3
g.SecurityModel = gosnmp.UserSecurityModel
usmSecurityParameters := gosnmp.UsmSecurityParameters{
@@ -162,12 +162,12 @@ func (p *realPoller) Poll(ctx context.Context, exporter string, port uint16, ifI
fmt.Printf("WWWWAA %+v\n", g)
} else {
g.Version = gosnmp.Version2c
g.Community = p.config.Communities.LookupOrDefault(exporterIP, "public")
g.Community = p.config.Communities.LookupOrDefault(exporter, "public")
}
if err := g.Connect(); err != nil {
p.metrics.failures.WithLabelValues(exporter, "connect").Inc()
p.errLogger.Err(err).Str("exporter", exporter).Msg("unable to connect")
p.metrics.failures.WithLabelValues(exporterStr, "connect").Inc()
p.errLogger.Err(err).Str("exporter", exporterStr).Msg("unable to connect")
}
start := p.clock.Now()
requests := []string{"1.3.6.1.2.1.1.5.0"}
@@ -184,17 +184,17 @@ func (p *realPoller) Poll(ctx context.Context, exporter string, port uint16, ifI
return nil
}
if err != nil {
p.metrics.failures.WithLabelValues(exporter, "get").Inc()
p.metrics.failures.WithLabelValues(exporterStr, "get").Inc()
p.errLogger.Err(err).
Str("exporter", exporter).
Str("exporter", exporterStr).
Msgf("unable to GET (%d OIDs)", len(requests))
return err
}
if result.Error != gosnmp.NoError && result.ErrorIndex == 0 {
// There is some error affecting the whole request
p.metrics.failures.WithLabelValues(exporter, "get").Inc()
p.metrics.failures.WithLabelValues(exporterStr, "get").Inc()
p.errLogger.Error().
Str("exporter", exporter).
Str("exporter", exporterStr).
Stringer("code", result.Error).
Msgf("unable to GET (%d OIDs)", len(requests))
return fmt.Errorf("SNMP error %s(%d)", result.Error, result.Error)
@@ -206,11 +206,11 @@ func (p *realPoller) Poll(ctx context.Context, exporter string, port uint16, ifI
*target = string(result.Variables[idx].Value.([]byte))
case gosnmp.NoSuchInstance, gosnmp.NoSuchObject:
if mandatory {
p.metrics.failures.WithLabelValues(exporter, fmt.Sprintf("%s missing", what)).Inc()
p.metrics.failures.WithLabelValues(exporterStr, fmt.Sprintf("%s missing", what)).Inc()
return false
}
default:
p.metrics.failures.WithLabelValues(exporter, fmt.Sprintf("%s unknown type", what)).Inc()
p.metrics.failures.WithLabelValues(exporterStr, fmt.Sprintf("%s unknown type", what)).Inc()
return false
}
return true
@@ -221,11 +221,11 @@ func (p *realPoller) Poll(ctx context.Context, exporter string, port uint16, ifI
*target = result.Variables[idx].Value.(uint)
case gosnmp.NoSuchInstance, gosnmp.NoSuchObject:
if mandatory {
p.metrics.failures.WithLabelValues(exporter, fmt.Sprintf("%s missing", what)).Inc()
p.metrics.failures.WithLabelValues(exporterStr, fmt.Sprintf("%s missing", what)).Inc()
return false
}
default:
p.metrics.failures.WithLabelValues(exporter, fmt.Sprintf("%s unknown type", what)).Inc()
p.metrics.failures.WithLabelValues(exporterStr, fmt.Sprintf("%s unknown type", what)).Inc()
return false
}
return true
@@ -259,10 +259,10 @@ func (p *realPoller) Poll(ctx context.Context, exporter string, port uint16, ifI
Description: ifAliasVal,
Speed: ifSpeedVal,
})
p.metrics.successes.WithLabelValues(exporter).Inc()
p.metrics.successes.WithLabelValues(exporterStr).Inc()
}
p.metrics.times.WithLabelValues(exporter).Observe(p.clock.Now().Sub(start).Seconds())
p.metrics.times.WithLabelValues(exporterStr).Observe(p.clock.Now().Sub(start).Seconds())
return nil
}

16
inlet/snmp/poller_test.go
View File

@@ -7,6 +7,7 @@ import (
"context"
"fmt"
"net"
"net/netip"
"strconv"
"testing"
"time"
@@ -83,8 +84,9 @@ func TestPoller(t *testing.T) {
r := reporter.NewMock(t)
clock := clock.NewMock()
config := tc.Config
p := newPoller(r, config, clock, func(exporterIP, exporterName string, ifIndex uint, iface Interface) {
got = append(got, fmt.Sprintf("%s %s %d %s %s %d", exporterIP, exporterName,
p := newPoller(r, config, clock, func(exporterIP netip.Addr, exporterName string, ifIndex uint, iface Interface) {
got = append(got, fmt.Sprintf("%s %s %d %s %s %d",
exporterIP.Unmap().String(), exporterName,
ifIndex, iface.Name, iface.Description, iface.Speed))
})
@@ -188,11 +190,11 @@ func TestPoller(t *testing.T) {
go server.ServeForever()
defer server.Shutdown()
p.Poll(context.Background(), "127.0.0.1", uint16(port), []uint{641})
p.Poll(context.Background(), "127.0.0.1", uint16(port), []uint{642})
p.Poll(context.Background(), "127.0.0.1", uint16(port), []uint{643})
p.Poll(context.Background(), "127.0.0.1", uint16(port), []uint{644})
p.Poll(context.Background(), "127.0.0.1", uint16(port), []uint{0})
p.Poll(context.Background(), netip.MustParseAddr("::ffff:127.0.0.1"), uint16(port), []uint{641})
p.Poll(context.Background(), netip.MustParseAddr("::ffff:127.0.0.1"), uint16(port), []uint{642})
p.Poll(context.Background(), netip.MustParseAddr("::ffff:127.0.0.1"), uint16(port), []uint{643})
p.Poll(context.Background(), netip.MustParseAddr("::ffff:127.0.0.1"), uint16(port), []uint{644})
p.Poll(context.Background(), netip.MustParseAddr("::ffff:127.0.0.1"), uint16(port), []uint{0})
time.Sleep(50 * time.Millisecond)
if diff := helpers.Diff(got, []string{
`127.0.0.1 exporter62 641 Gi0/0/0/0 Transit 10000`,

26
inlet/snmp/root.go
View File

@@ -9,7 +9,7 @@ package snmp
import (
"errors"
"fmt"
"net"
"net/netip"
"strconv"
"sync"
"time"
@@ -36,8 +36,8 @@ type Component struct {
dispatcherChannel chan lookupRequest
dispatcherBChannel chan (<-chan bool) // block channel for testing
pollerBreakersLock sync.Mutex
pollerBreakerLoggers map[string]reporter.Logger
pollerBreakers map[string]*breaker.Breaker
pollerBreakerLoggers map[netip.Addr]reporter.Logger
pollerBreakers map[netip.Addr]*breaker.Breaker
poller poller
metrics struct {
@@ -76,8 +76,8 @@ func New(r *reporter.Reporter, configuration Configuration, dependencies Depende
pollerChannel: make(chan lookupRequest),
dispatcherChannel: make(chan lookupRequest, 100*configuration.Workers),
dispatcherBChannel: make(chan (<-chan bool)),
pollerBreakers: make(map[string]*breaker.Breaker),
pollerBreakerLoggers: make(map[string]reporter.Logger),
pollerBreakers: make(map[netip.Addr]*breaker.Breaker),
pollerBreakerLoggers: make(map[netip.Addr]reporter.Logger),
poller: newPoller(r, pollerConfig{
Retries: configuration.PollerRetries,
Timeout: configuration.PollerTimeout,
@@ -218,14 +218,14 @@ func (c *Component) Stop() error {
// lookupRequest is used internally to queue a polling request.
type lookupRequest struct {
ExporterIP string
ExporterIP netip.Addr
IfIndexes []uint
}
// Lookup for interface information for the provided exporter and ifIndex.
// If the information is not in the cache, it will be polled, but
// won't be returned immediately.
func (c *Component) Lookup(exporterIP string, ifIndex uint) (string, Interface, error) {
func (c *Component) Lookup(exporterIP netip.Addr, ifIndex uint) (string, Interface, error) {
exporterName, iface, err := c.sc.Lookup(exporterIP, ifIndex)
if errors.Is(err, ErrCacheMiss) {
req := lookupRequest{
@@ -235,7 +235,7 @@ func (c *Component) Lookup(exporterIP string, ifIndex uint) (string, Interface,
select {
case c.dispatcherChannel <- req:
default:
c.metrics.pollerBusyCount.WithLabelValues(exporterIP).Inc()
c.metrics.pollerBusyCount.WithLabelValues(exporterIP.Unmap().String()).Inc()
}
}
return exporterName, iface, err
@@ -244,7 +244,7 @@ func (c *Component) Lookup(exporterIP string, ifIndex uint) (string, Interface,
// Dispatch an incoming request to workers. May handle more than the
// provided request if it can.
func (c *Component) dispatchIncomingRequest(request lookupRequest) {
requestsMap := map[string][]uint{
requestsMap := map[netip.Addr][]uint{
request.ExporterIP: request.IfIndexes,
}
for c.config.PollerCoalesce > 0 {
@@ -298,16 +298,16 @@ func (c *Component) pollerIncomingRequest(request lookupRequest) {
return c.poller.Poll(
c.t.Context(nil),
request.ExporterIP,
c.config.Ports.LookupOrDefault(net.ParseIP(request.ExporterIP), 161),
c.config.Ports.LookupOrDefault(request.ExporterIP, 161),
request.IfIndexes)
}); err == breaker.ErrBreakerOpen {
c.metrics.pollerBreakerOpenCount.WithLabelValues(request.ExporterIP).Inc()
c.metrics.pollerBreakerOpenCount.WithLabelValues(request.ExporterIP.Unmap().String()).Inc()
c.pollerBreakersLock.Lock()
l, ok := c.pollerBreakerLoggers[request.ExporterIP]
if !ok {
l = c.r.Sample(reporter.BurstSampler(time.Minute, 1)).
With().
Str("exporter", request.ExporterIP).
Str("exporter", request.ExporterIP.Unmap().String()).
Logger()
c.pollerBreakerLoggers[request.ExporterIP] = l
}
@@ -333,7 +333,7 @@ func (c *Component) expireCache() {
}:
count++
default:
c.metrics.pollerBusyCount.WithLabelValues(exporter).Inc()
c.metrics.pollerBusyCount.WithLabelValues(exporter.Unmap().String()).Inc()
}
}
}

18
inlet/snmp/root_test.go
View File

@@ -6,6 +6,7 @@ package snmp
import (
"context"
"errors"
"net/netip"
"path/filepath"
"testing"
"time"
@@ -19,7 +20,8 @@ import (
func expectSNMPLookup(t *testing.T, c *Component, exporter string, ifIndex uint, expected answer) {
t.Helper()
gotExporterName, gotInterface, err := c.Lookup(exporter, ifIndex)
ip := netip.AddrFrom16(netip.MustParseAddr(exporter).As16())
gotExporterName, gotInterface, err := c.Lookup(ip, ifIndex)
got := answer{gotExporterName, gotInterface, err}
if diff := helpers.Diff(got, expected); diff != "" {
t.Fatalf("Lookup() (-got, +want):\n%s", diff)
@@ -109,9 +111,9 @@ func TestAutoRefresh(t *testing.T) {
// Keep it in the cache!
mockClock.Add(25 * time.Minute)
c.Lookup("127.0.0.1", 765)
c.Lookup(netip.MustParseAddr("::ffff:127.0.0.1"), 765)
mockClock.Add(25 * time.Minute)
c.Lookup("127.0.0.1", 765)
c.Lookup(netip.MustParseAddr("::ffff:127.0.0.1"), 765)
// Go forward, we expect the entry to have been refreshed and be still present
mockClock.Add(11 * time.Minute)
@@ -179,7 +181,7 @@ type logCoalescePoller struct {
received []lookupRequest
}
func (fcp *logCoalescePoller) Poll(ctx context.Context, exporterIP string, _ uint16, ifIndexes []uint) error {
func (fcp *logCoalescePoller) Poll(ctx context.Context, exporterIP netip.Addr, _ uint16, ifIndexes []uint) error {
fcp.received = append(fcp.received, lookupRequest{exporterIP, ifIndexes})
return nil
}
@@ -218,7 +220,7 @@ func TestCoalescing(t *testing.T) {
}
expectedAccepted := []lookupRequest{
{"127.0.0.1", []uint{766, 767, 768, 769}},
{netip.MustParseAddr("::ffff:127.0.0.1"), []uint{766, 767, 768, 769}},
}
if diff := helpers.Diff(lcp.received, expectedAccepted); diff != "" {
t.Errorf("Accepted requests (-got, +want):\n%s", diff)
@@ -227,7 +229,7 @@ func TestCoalescing(t *testing.T) {
type errorPoller struct{}
func (fcp *errorPoller) Poll(ctx context.Context, exporterIP string, _ uint16, ifIndexes []uint) error {
func (fcp *errorPoller) Poll(ctx context.Context, exporterIP netip.Addr, _ uint16, ifIndexes []uint) error {
return errors.New("noooo")
}
@@ -252,10 +254,10 @@ func TestPollerBreaker(t *testing.T) {
c.metrics.pollerBreakerOpenCount.WithLabelValues("127.0.0.1").Add(0)
for i := 0; i < 30; i++ {
c.Lookup("127.0.0.1", 765)
c.Lookup(netip.MustParseAddr("::ffff:127.0.0.1"), 765)
}
for i := 0; i < 5; i++ {
c.Lookup("127.0.0.2", 765)
c.Lookup(netip.MustParseAddr("::ffff:127.0.0.2"), 765)
}
time.Sleep(50 * time.Millisecond)

12
inlet/snmp/tests.go
View File

@@ -8,7 +8,7 @@ package snmp
import (
"context"
"fmt"
"net"
"net/netip"
"strings"
"testing"
@@ -19,11 +19,11 @@ import (
// mockPoller will use static data.
type mockPoller struct {
config Configuration
put func(string, string, uint, Interface)
put func(netip.Addr, string, uint, Interface)
}
// newMockPoller creates a fake SNMP poller.
func newMockPoller(configuration Configuration, put func(string, string, uint, Interface)) *mockPoller {
func newMockPoller(configuration Configuration, put func(netip.Addr, string, uint, Interface)) *mockPoller {
return &mockPoller{
config: configuration,
put: put,
@@ -31,10 +31,10 @@ func newMockPoller(configuration Configuration, put func(string, string, uint, I
}
// Poll just builds synthetic data.
func (p *mockPoller) Poll(ctx context.Context, exporter string, port uint16, ifIndexes []uint) error {
func (p *mockPoller) Poll(ctx context.Context, exporter netip.Addr, port uint16, ifIndexes []uint) error {
for _, ifIndex := range ifIndexes {
if p.config.Communities.LookupOrDefault(net.ParseIP(exporter), "public") == "public" {
p.put(exporter, strings.ReplaceAll(exporter, ".", "_"), ifIndex, Interface{
if p.config.Communities.LookupOrDefault(exporter, "public") == "public" {
p.put(exporter, strings.ReplaceAll(exporter.Unmap().String(), ".", "_"), ifIndex, Interface{
Name: fmt.Sprintf("Gi0/0/%d", ifIndex),
Description: fmt.Sprintf("Interface %d", ifIndex),
Speed: 1000,