Repository: kentik/ktranslate
Pull request #319 (2022-04-28 01:38:57)

Adding a list oriented match for profile
Goes with  #317. This allows deterministic ordering of sysdesc profiles to match on. 


Patch:
diff --git a/deployment/kubernetes/ktranslate-nr.yaml b/deployment/kubernetes/ktranslate-nr.yaml
new file mode 100644
index 0000000..08eb144
--- /dev/null
+++ b/deployment/kubernetes/ktranslate-nr.yaml
@@ -0,0 +1,136 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: ktranslate
+spec:
+  selector:
+    matchLabels:
+      app: ktranslate
+  template:
+    metadata:
+      labels:
+        app: ktranslate
+    spec:
+      containers:
+      - name: ktranslate
+        image: docker.io/kentik/ktranslate:v2
+        imagePullPolicy: Always
+        env:
+        - name: NEW_RELIC_API_KEY
+          valueFrom:
+            secretKeyRef:
+              name: ktranslate-secret
+              key: nr_api_key
+        - name: NR_ACCOUNT_ID
+          valueFrom:
+            secretKeyRef:
+              name: ktranslate-secret
+              key: nr_account_id
+        args:
+          - --metalisten=0.0.0.0:8083
+          - --snmp=/etc/ktranslate/snmp.yml
+          - --metrics=jchf
+          - --tee_logs=true
+          - nr1.snmp
+        resources:
+          limits:
+            cpu: 1000m
+            memory: 256Mi
+          requests:
+            cpu: 500m
+            memory: 128Mi
+        ports:
+        - name: metadata
+          containerPort: 8083
+        - name: snmp
+          containerPort: 1620
+          protocol: UDP
+        livenessProbe:
+          httpGet:
+            path: /metrics
+            port: metadata
+          initialDelaySeconds: 5
+          periodSeconds: 5
+        readinessProbe:
+          httpGet:
+            path: /metrics
+            port: metadata
+          initialDelaySeconds: 5
+          periodSeconds: 5
+        volumeMounts:
+          - name: ktranslate-config
+            mountPath: /etc/ktranslate/snmp.yml
+            subPath: snmp.yml
+      volumes:
+        - name: ktranslate-config
+          configMap:
+            name: ktranslate-config
+
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: ktranslate-metadata
+  labels:
+    app: ktranslate
+spec:
+  ports:
+  - port: 8083
+    name: metadata
+    targetPort: 8083
+    protocol: TCP
+  selector:
+    app: ktranslate
+
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: ktranslate-snmp
+  labels:
+    app: ktranslate
+spec:
+  ports:
+  - port: 1620
+    name: snmp
+    targetPort: 1620
+    protocol: UDP
+  selector:
+    app: ktranslate
+
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: ktranslate-config
+data:
+  snmp.yml: |
+    devices:
+      switch:
+        device_name: switch
+        device_ip: 10.10.0.10
+        flow_only: true
+        user_tags: {}
+    trap:
+      listen: 0.0.0.0:1620
+      community: hello
+      version: ""
+      transport: ""
+    global:
+      poll_time_sec: 30
+      drop_if_outside_poll: false
+      mib_profile_dir: /etc/ktranslate/profiles
+      mibs_db: /etc/ktranslate/mibs.db
+      mibs_enabled:
+      - IF-MIB
+      timeout_ms: 3000
+      retries: 0
+
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  name: ktranslate-secret
+data:
+  nr_account_id: bXktYXBw
+  nr_api_key: Mzk1MjgkdmRnN0pi
\ No newline at end of file
diff --git a/pkg/cat/kkc.go b/pkg/cat/kkc.go
index 3d9397a..c638994 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
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/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


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 {