Repository: kentik/ktranslate
Pull request #321 (2022-04-28 22:37:44)

Resolve flow sending devices by dns
If a device's IP is not in `snmp.yaml`, netflow collector doesn't know what to call it. This will use dns if possible to resolve it to a device name.

Also adds locking to the resolver class which probably should have been there the whole time. 

#315


Patch:
diff --git a/pkg/cat/jchf.go b/pkg/cat/jchf.go
index 300a3d9..afd0eb4 100644
--- a/pkg/cat/jchf.go
+++ b/pkg/cat/jchf.go
@@ -248,7 +248,7 @@ func (kc *KTranslate) flowToJCHF(ctx context.Context, citycache map[uint32]strin
 
 	// Resolve any hostnames if a resolver is set up.
 	if kc.resolver != nil {
-		dst.CustomStr["dst_host"] = kc.resolver.Resolve(ctx, dst.DstAddr)
+		dst.CustomStr["dst_host"] = kc.resolver.Resolve(ctx, dst.DstAddr, false)
 	}
 
 	if src.CHF.Ipv4SrcAddr() > 0 {
@@ -266,7 +266,7 @@ func (kc *KTranslate) flowToJCHF(ctx context.Context, citycache map[uint32]strin
 	dst.DstRoutePrefix = addr.String()
 
 	if kc.resolver != nil {
-		dst.CustomStr["src_host"] = kc.resolver.Resolve(ctx, dst.SrcAddr)
+		dst.CustomStr["src_host"] = kc.resolver.Resolve(ctx, dst.SrcAddr, false)
 	}
 
 	// next hops
@@ -557,8 +557,8 @@ func (kc *KTranslate) doEnrichments(ctx context.Context, citycache map[uint32]st
 
 		// If there's a resolver, try to resolve to hostnames.
 		if kc.resolver != nil {
-			msg.CustomStr["src_host"] = kc.resolver.Resolve(ctx, msg.SrcAddr)
-			msg.CustomStr["dst_host"] = kc.resolver.Resolve(ctx, msg.DstAddr)
+			msg.CustomStr["src_host"] = kc.resolver.Resolve(ctx, msg.SrcAddr, false)
+			msg.CustomStr["dst_host"] = kc.resolver.Resolve(ctx, msg.DstAddr, false)
 		}
 	}
 
diff --git a/pkg/cat/kkc.go b/pkg/cat/kkc.go
index 3d9397a..d764836 100644
--- a/pkg/cat/kkc.go
+++ b/pkg/cat/kkc.go
@@ -65,6 +65,7 @@ func NewKTranslate(config *Config, log logger.ContextL, registry go_metrics.Regi
 			JCHFQ:        go_metrics.GetOrRegisterGauge("jchfq", registry),
 			InputQ:       go_metrics.GetOrRegisterMeter("inputq", registry),
 			InputQLen:    go_metrics.GetOrRegisterGauge("inputq_len^force=true", registry),
+			OutputQLen:   go_metrics.GetOrRegisterGauge("outputq_len^force=true", registry),
 		},
 		alphaChans: make([]chan *Flow, config.Threads),
 		jchfChans:  make([]chan *kt.JCHF, config.Threads),
@@ -254,6 +255,7 @@ func (kc *KTranslate) HttpInfo(w http.ResponseWriter, r *http.Request) {
 		AlphaQDrop:     kc.metrics.AlphaQDrop.Rate1(),
 		InputQ:         kc.metrics.InputQ.Rate1(),
 		InputQLen:      kc.metrics.InputQLen.Value(),
+		OutputQLen:     kc.metrics.OutputQLen.Value(),
 		Sinks:          map[ss.Sink]map[string]float64{},
 		SnmpDeviceData: map[string]map[string]float64{},
 		Inputs:         map[string]map[string]float64{},
@@ -452,6 +454,7 @@ func (kc *KTranslate) watchInput(ctx context.Context, seri func([]*kt.JCHF, []by
 				}
 			}
 			kc.metrics.InputQLen.Update(int64(len(kc.inputChan)))
+			kc.metrics.OutputQLen.Update(int64(len(kc.msgsc)))
 		case <-ctx.Done():
 			kc.log.Infof("watchInput Done")
 			return
@@ -656,7 +659,7 @@ func (kc *KTranslate) Run(ctx context.Context) error {
 	// If we're looking for netflow direct flows coming in
 	if kc.config.FlowSource != "" {
 		assureInput()
-		nfs, err := flow.NewFlowSource(ctx, kc.config.FlowSource, kc.config.MaxFlowPerMessage, kc.log.GetLogger().GetUnderlyingLogger(), kc.registry, kc.inputChan, kc.apic)
+		nfs, err := flow.NewFlowSource(ctx, kc.config.FlowSource, kc.config.MaxFlowPerMessage, kc.log.GetLogger().GetUnderlyingLogger(), kc.registry, kc.inputChan, kc.apic, kc.resolver)
 		if err != nil {
 			return err
 		}
diff --git a/pkg/cat/types.go b/pkg/cat/types.go
index 7ab1314..329df20 100644
--- a/pkg/cat/types.go
+++ b/pkg/cat/types.go
@@ -137,6 +137,7 @@ type hc struct {
 	AlphaQDrop     float64
 	InputQ         float64
 	InputQLen      int64
+	OutputQLen     int64
 	Sinks          map[sinks.Sink]map[string]float64
 	SnmpDeviceData map[string]map[string]float64
 	Inputs         map[string]map[string]float64
@@ -157,5 +158,6 @@ type KKCMetric struct {
 	AlphaQDrop     go_metrics.Meter
 	InputQ         go_metrics.Meter
 	InputQLen      go_metrics.Gauge
+	OutputQLen     go_metrics.Gauge
 	SnmpDeviceData *kt.SnmpMetricSet
 }
diff --git a/pkg/inputs/flow/flow.go b/pkg/inputs/flow/flow.go
index fa64bac..b11bbd2 100644
--- a/pkg/inputs/flow/flow.go
+++ b/pkg/inputs/flow/flow.go
@@ -12,6 +12,7 @@ import (
 	"github.com/kentik/ktranslate/pkg/api"
 	"github.com/kentik/ktranslate/pkg/eggs/logger"
 	"github.com/kentik/ktranslate/pkg/kt"
+	"github.com/kentik/ktranslate/pkg/util/resolv"
 
 	"github.com/netsampler/goflow2/utils"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -40,8 +41,8 @@ var (
 	MappingFile   = flag.String("nf.mapping", "", "Configuration file for custom netflow mappings")
 )
 
-func NewFlowSource(ctx context.Context, proto FlowSource, maxBatchSize int, log logger.Underlying, registry go_metrics.Registry, jchfChan chan []*kt.JCHF, apic *api.KentikApi) (*KentikDriver, error) {
-	kt := NewKentikDriver(ctx, proto, maxBatchSize, log, registry, jchfChan, apic, *MessageFields)
+func NewFlowSource(ctx context.Context, proto FlowSource, maxBatchSize int, log logger.Underlying, registry go_metrics.Registry, jchfChan chan []*kt.JCHF, apic *api.KentikApi, resolv *resolv.Resolver) (*KentikDriver, error) {
+	kt := NewKentikDriver(ctx, proto, maxBatchSize, log, registry, jchfChan, apic, *MessageFields, resolv)
 	kt.Infof("Netflow listener running on %s:%d for format %s and a batch size of %d", *Addr, *Port, proto, maxBatchSize)
 	kt.Infof("Netflow listener sending fields %s", *MessageFields)
 
diff --git a/pkg/inputs/flow/format.go b/pkg/inputs/flow/format.go
index 59ab219..ddf561c 100644
--- a/pkg/inputs/flow/format.go
+++ b/pkg/inputs/flow/format.go
@@ -15,6 +15,7 @@ import (
 	"github.com/kentik/ktranslate/pkg/eggs/logger"
 	"github.com/kentik/ktranslate/pkg/kt"
 	"github.com/kentik/ktranslate/pkg/util/ic"
+	"github.com/kentik/ktranslate/pkg/util/resolv"
 
 	flowmessage "github.com/netsampler/goflow2/pb"
 )
@@ -34,13 +35,15 @@ type KentikDriver struct {
 	inputs       chan *kt.JCHF
 	proto        FlowSource
 	registry     go_metrics.Registry
+	resolv       *resolv.Resolver
+	ctx          context.Context
 }
 
 type FlowMetric struct {
 	Flows go_metrics.Meter
 }
 
-func NewKentikDriver(ctx context.Context, proto FlowSource, maxBatchSize int, log logger.Underlying, registry go_metrics.Registry, jchfChan chan []*kt.JCHF, apic *api.KentikApi, fields string) *KentikDriver {
+func NewKentikDriver(ctx context.Context, proto FlowSource, maxBatchSize int, log logger.Underlying, registry go_metrics.Registry, jchfChan chan []*kt.JCHF, apic *api.KentikApi, fields string, resolv *resolv.Resolver) *KentikDriver {
 	kt := KentikDriver{
 		ContextL:     logger.NewContextLFromUnderlying(logger.SContext{S: "flow"}, log),
 		jchfChan:     jchfChan,
@@ -52,6 +55,8 @@ func NewKentikDriver(ctx context.Context, proto FlowSource, maxBatchSize int, lo
 		inputs:       make(chan *kt.JCHF, maxBatchSize),
 		proto:        proto,
 		registry:     registry,
+		resolv:       resolv,
+		ctx:          ctx,
 	}
 	go kt.run(ctx) // Process flows and send them on.
 	return &kt
@@ -106,6 +111,12 @@ func (t *KentikDriver) toJCHF(fmsg *flowmessage.FlowMessage) *kt.JCHF {
 		dev.SetUserTags(in.CustomStr)
 	} else {
 		in.DeviceName = net.IP(fmsg.SamplerAddress).String()
+		if t.resolv != nil {
+			dm := t.resolv.Resolve(t.ctx, in.DeviceName, true)
+			if dm != "" {
+				in.DeviceName = dm
+			}
+		}
 	}
 
 	if _, ok := t.metrics[in.DeviceName]; !ok {
diff --git a/pkg/inputs/snmp/disco.go b/pkg/inputs/snmp/disco.go
index a5894fb..d07e483 100644
--- a/pkg/inputs/snmp/disco.go
+++ b/pkg/inputs/snmp/disco.go
@@ -139,25 +139,34 @@ func Discover(ctx context.Context, snmpFile string, log logger.ContextL) (*SnmpD
 	return stats, nil
 }
 
-func RunDiscoOnTimer(ctx context.Context, c chan os.Signal, snmpFile string, log logger.ContextL, pollTimeMin int) {
+func RunDiscoOnTimer(ctx context.Context, c chan os.Signal, snmpFile string, log logger.ContextL, pollTimeMin int, checkNow bool) {
 	pt := time.Duration(pollTimeMin) * time.Minute
 	log.Infof("Running SNMP Discovery Loop every %v", pt)
 	discoCheck := time.NewTicker(pt)
 	defer discoCheck.Stop()
+
+	check := func() {
+		stats, err := Discover(ctx, snmpFile, log)
+		if err != nil {
+			log.Errorf("Discovery SNMP Error: %v", err)
+		} else {
+			if stats.delta != 0 || stats.added > 0 { // Only restart if there's a different configuration.
+				log.Infof("Discovery SNMP reloading: added: %d replaced: %d delta: %d", stats.added, stats.replaced, stats.delta)
+				c <- kt.SIGUSR2 // Restart the main loop with a new config.
+			} else {
+				log.Infof("Discovery SNMP no change so not reloading: added: %d replaced: %d delta: %d", stats.added, stats.replaced, stats.delta)
+			}
+		}
+	}
+
+	if checkNow {
+		check()
+	}
+
 	for {
 		select {
 		case _ = <-discoCheck.C:
-			stats, err := Discover(ctx, snmpFile, log)
-			if err != nil {
-				log.Errorf("Discovery SNMP Error: %v", err)
-			} else {
-				if stats.delta != 0 || stats.added > 0 { // Only restart if there's a different configuration.
-					log.Infof("Discovery SNMP reloading: added: %d replaced: %d delta: %d", stats.added, stats.replaced, stats.delta)
-					c <- kt.SIGUSR2 // Restart the main loop with a new config.
-				} else {
-					log.Infof("Discovery SNMP no change so not reloading: added: %d replaced: %d delta: %d", stats.added, stats.replaced, stats.delta)
-				}
-			}
+			check()
 		case <-ctx.Done():
 			log.Infof("Discovery Loop Done")
 			return
diff --git a/pkg/inputs/snmp/mibs/profile.go b/pkg/inputs/snmp/mibs/profile.go
index 3bb6d3b..d9acf1c 100644
--- a/pkg/inputs/snmp/mibs/profile.go
+++ b/pkg/inputs/snmp/mibs/profile.go
@@ -52,6 +52,11 @@ type Device struct {
 	Vendor string `yaml:"vendor,omitempty"`
 }
 
+type Match struct {
+	Regex  string `yaml:"regex"`
+	Target string `yaml:"target"`
+}
+
 type Profile struct {
 	logger.ContextL  `yaml:"-"`
 	Metrics          []MIB             `yaml:"metrics,omitempty"`
@@ -63,6 +68,7 @@ type Profile struct {
 	Provider         kt.Provider       `yaml:"provider,omitempty"`
 	NoUseBulkWalkAll bool              `yaml:"no_use_bulkwalkall"`
 	Matches          map[string]string `yaml:"matches"`
+	MatchesList      []Match           `yaml:"matches_list"`
 	SysMap           map[string]string
 	extended         bool
 }
@@ -261,6 +267,24 @@ func (mdb *MibDB) FindProfile(sysid string, sysdesc string, mibProfile string) *
 }
 
 func (mdb *MibDB) checkMatch(pro *Profile, sysdesc string) *Profile {
+	// Match on list first because this is ordered.
+	for _, match := range pro.MatchesList {
+		r, err := regexp.Compile(match.Regex)
+		if err != nil {
+			mdb.log.Errorf("Invalid Regex for Match: %s %s", pro.From, match.Regex)
+			return nil
+		}
+		if r.MatchString(strings.ToLower(sysdesc)) {
+			for _, p := range mdb.profiles {
+				if p.From == match.Target {
+					return p
+				}
+			}
+			mdb.log.Errorf("No profile matching %s found", match.Target)
+		}
+	}
+
+	// And then the old way, which randomly itterates over the matches map.
 	for m, target := range pro.Matches {
 		r, err := regexp.Compile(m)
 		if err != nil {
diff --git a/pkg/inputs/snmp/snmp.go b/pkg/inputs/snmp/snmp.go
index 023a5ee..f5b2a92 100644
--- a/pkg/inputs/snmp/snmp.go
+++ b/pkg/inputs/snmp/snmp.go
@@ -34,6 +34,7 @@ var (
 	snmpOutFile    = flag.String("snmp_out_file", "", "If set, write updated snmp file here.")
 	snmpPollNow    = flag.String("snmp_poll_now", "", "If set, run one snmp poll for the specified device and then exit.")
 	snmpDiscoDur   = flag.Int("snmp_discovery_min", 0, "If set, run snmp discovery on this interval (in minutes).")
+	snmpDiscoSt    = flag.Bool("snmp_discovery_on_start", false, "If set, run snmp discovery on application start.")
 	validateMib    = flag.Bool("snmp_validate", false, "If true, validate mib profiles and exit.")
 	ServiceName    = ""
 )
@@ -135,7 +136,7 @@ func wrapSnmpPolling(ctx context.Context, snmpFile string, jchfChan chan []*kt.J
 	c := make(chan os.Signal, 1)
 	signal.Notify(c, kt.SIGUSR2)
 	if *snmpDiscoDur > 0 { // If we are re-running snmp discovery every interval, start the ticker here.
-		go RunDiscoOnTimer(ctxSnmp, c, snmpFile, log, *snmpDiscoDur)
+		go RunDiscoOnTimer(ctxSnmp, c, snmpFile, log, *snmpDiscoDur, *snmpDiscoSt)
 	}
 
 	// Block here
diff --git a/pkg/inputs/syslog/syslog.go b/pkg/inputs/syslog/syslog.go
index 87f41a9..35e1852 100644
--- a/pkg/inputs/syslog/syslog.go
+++ b/pkg/inputs/syslog/syslog.go
@@ -188,7 +188,7 @@ func (ks *KentikSyslog) formatMessage(ctx context.Context, msg sfmt.LogParts) ([
 			dev.SetMsgUserTags(msg)
 		}
 		if ks.resolver != nil {
-			msg["client_name"] = ks.resolver.Resolve(ctx, pts[0])
+			msg["client_name"] = ks.resolver.Resolve(ctx, pts[0], true)
 		}
 	}
 
@@ -198,7 +198,7 @@ func (ks *KentikSyslog) formatMessage(ctx context.Context, msg sfmt.LogParts) ([
 			if hs, ok := hostname.(string); ok {
 				if ipr := net.ParseIP(hs); ipr != nil {
 					if ks.resolver != nil {
-						msg["device_name"] = ks.resolver.Resolve(ctx, hs)
+						msg["device_name"] = ks.resolver.Resolve(ctx, hs, true)
 					} else {
 						msg["device_name"] = hs
 					}
diff --git a/pkg/util/resolv/resolver.go b/pkg/util/resolv/resolver.go
index b60d73c..284631d 100644
--- a/pkg/util/resolv/resolver.go
+++ b/pkg/util/resolv/resolver.go
@@ -6,6 +6,7 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/kentik/ktranslate/pkg/eggs/logger"
@@ -25,6 +26,7 @@ type Resolver struct {
 	logger.ContextL
 	resolver *net.Resolver
 	cache    map[string]string
+	mux      sync.RWMutex
 }
 
 func NewResolver(ctx context.Context, log logger.Underlying, dnsHost string) (*Resolver, error) {
@@ -64,11 +66,14 @@ func NewResolver(ctx context.Context, log logger.Underlying, dnsHost string) (*R
 	return res, nil
 }
 
-func (r *Resolver) Resolve(ctx context.Context, ip string) string {
+func (r *Resolver) Resolve(ctx context.Context, ip string, log bool) string {
+	r.mux.RLock()
 	final, ok := r.cache[ip]
 	if ok {
+		r.mux.RUnlock()
 		return final
 	}
+	r.mux.RUnlock()
 
 	// Else, look it up on the network, unless we are full.
 	if len(r.cache) > MAX_CACHE_SIZE {
@@ -87,7 +92,12 @@ func (r *Resolver) Resolve(ctx context.Context, ip string) string {
 			}
 		}
 	} // ignore errors here
+	r.mux.Lock()
 	r.cache[ip] = final // cache.
+	r.mux.Unlock()
+	if log {
+		r.Infof("%s resolved to %s", ip, final)
+	}
 
 	return final
 }


Test patch:
diff --git a/pkg/inputs/snmp/mibs/profile_test.go b/pkg/inputs/snmp/mibs/profile_test.go
index 7dc557f..d17111b 100644
--- a/pkg/inputs/snmp/mibs/profile_test.go
+++ b/pkg/inputs/snmp/mibs/profile_test.go
@@ -26,7 +26,10 @@ func TestFindProfile(t *testing.T) {
 			"1.3.6.1.4.1.318.1.3.*":       &Profile{Device: Device{Vendor: "ups"}},
 			"1.3.6.1.4.1.8072.3.2.*":      &Profile{Matches: map[string]string{"^barracuda": "barracuda.yml"}, Device: Device{Vendor: "net-bsd"}},
 			"1.3.6.1.4.1.1588.2.1.1.*":    &Profile{From: "barracuda.yml", Device: Device{Vendor: "barracuda"}},
+			"1.3.6.1.4.1.1588.2.1.22.*":   &Profile{From: "foo.yml", Device: Device{Vendor: "foo"}},
 			"1.3.6.1.4.1.2.3.4":           &Profile{From: "direct", Device: Device{Vendor: "direct"}},
+			"1.3.6.1.4.1.8072.3.66.*":     &Profile{MatchesList: []Match{Match{Regex: "^barracuda", Target: "barracuda.yml"}}, Device: Device{Vendor: "net-bsd"}},
+			"1.3.6.1.4.1.8072.3.77.*":     &Profile{MatchesList: []Match{Match{Regex: "^foocuda", Target: "foo.yml"}}, Matches: map[string]string{"^barracuda": "barracuda.yml"}, Device: Device{Vendor: "net-bsd"}},
 		},
 	}
 
@@ -104,6 +107,16 @@ func TestFindProfile(t *testing.T) {
 			Sysdesc:  "Barracuda Email Security Gateway",
 			Profile:  "!direct",
 		},
+		{
+			Sysoid:   ".1.3.6.1.4.1.8072.3.66.10",
+			Expected: "barracuda",
+			Sysdesc:  "Barracuda Email Security Gateway",
+		},
+		{
+			Sysoid:   ".1.3.6.1.4.1.8072.3.77.10",
+			Expected: "foo",
+			Sysdesc:  "Foocuda Email Security Gateway",
+		},
 	}
 
 	for _, in := range inputs {