Repository: kentik/ktranslate
Pull request #356 (2022-07-12 22:02:45)

Refactor influx
This PR extracts some general format stuff from nrm and makes it available to any format. Specifically, the code which keeps values from one mib from being set for all mibs is now present, limiting attributes to only those which make sense. 


Patch:
diff --git a/.github/workflows/publish-prod.yml b/.github/workflows/publish-prod.yml
index ea6f23e..b4d4231 100644
--- a/.github/workflows/publish-prod.yml
+++ b/.github/workflows/publish-prod.yml
@@ -8,6 +8,9 @@ on:
     branches:
       - main
 
+  release:
+    types: [created]
+
 jobs:
   deploy:
     runs-on: ubuntu-latest
@@ -17,9 +20,20 @@ jobs:
           submodules: recursive
           fetch-depth: 0         # Fetch all history for .GitInfo and .Lastmod
 
-      - name: Set Version
+      - uses: nowsprinting/check-version-format-action@v3
+        id: version
+        with:
+          prefix: 'v'
+
+      - name: Set Release Version
+        run: |
+           echo "KENTIK_KTRANSLATE_VERSION=${GITHUB_REF#refs/*/}" >> $GITHUB_ENV
+        if: steps.version.outputs.is_valid == 'true'
+
+      - name: Set NonRelease Version
         run: |
            echo KENTIK_KTRANSLATE_VERSION=`date +"kt-%Y-%m-%d-${GITHUB_RUN_ID}"` >> $GITHUB_ENV
+        if: steps.version.outputs.is_valid == 'false'
 
       - name: Setup Go
         uses: actions/setup-go@v2
diff --git a/pkg/cat/kkc.go b/pkg/cat/kkc.go
index d764836..ab5a879 100644
--- a/pkg/cat/kkc.go
+++ b/pkg/cat/kkc.go
@@ -629,7 +629,7 @@ func (kc *KTranslate) Run(ctx context.Context) error {
 	// If SNMP is configured, start this system too. Poll for metrics and metadata, also handle traps.
 	if kc.config.SNMPFile != "" {
 		if kc.config.SNMPDisco { // Here, we're just returning the list of devices on the network which might speak snmp.
-			_, err := snmp.Discover(ctx, kc.config.SNMPFile, kc.log)
+			_, err := snmp.Discover(ctx, kc.config.SNMPFile, kc.log, 0)
 			return err
 		}
 		assureInput()
diff --git a/pkg/formats/influx/influx.go b/pkg/formats/influx/influx.go
index a5d95e9..dd76d78 100644
--- a/pkg/formats/influx/influx.go
+++ b/pkg/formats/influx/influx.go
@@ -225,19 +225,39 @@ func (f *InfluxFormat) fromSnmpDeviceMetric(in *kt.JCHF) []InfluxData {
 	f.mux.RLock()
 	util.SetAttr(attr, in, metrics, f.lastMetadata[in.DeviceName])
 	f.mux.RUnlock()
-	ms := map[string]int64{}
-	for m, _ := range metrics {
+
+	results := []InfluxData{}
+	for m, name := range metrics {
+		if m == "" {
+			f.Errorf("Missing metric name, skipping %v", attr)
+			continue
+		}
 		if _, ok := in.CustomBigInt[m]; ok {
-			ms[m] = in.CustomBigInt[m]
+			attrNew := util.CopyAttrForSnmp(attr, m, name, f.lastMetadata[in.DeviceName])
+			if util.DropOnFilter(attrNew, f.lastMetadata[in.DeviceName], false) {
+				continue // This Metric isn't in the white list so lets drop it.
+			}
+
+			mtype := name.GetType()
+			if name.Format == kt.FloatMS {
+				results = append(results, InfluxData{
+					Name:        "kentik." + mtype,
+					FieldsFloat: map[string]float64{m: float64(float64(in.CustomBigInt[m]) / 1000)},
+					Timestamp:   in.Timestamp * 1000000000,
+					Tags:        attrNew,
+				})
+			} else {
+				results = append(results, InfluxData{
+					Name:      "kentik." + mtype,
+					Fields:    map[string]int64{m: int64(in.CustomBigInt[m])},
+					Timestamp: in.Timestamp * 1000000000,
+					Tags:      attrNew,
+				})
+			}
 		}
 	}
 
-	return []InfluxData{InfluxData{
-		Name:      "kentik.snmp.device",
-		Fields:    ms,
-		Timestamp: in.Timestamp * 1000000000,
-		Tags:      attr,
-	}}
+	return results
 }
 
 func (f *InfluxFormat) fromSnmpInterfaceMetric(in *kt.JCHF) []InfluxData {
@@ -246,11 +266,37 @@ func (f *InfluxFormat) fromSnmpInterfaceMetric(in *kt.JCHF) []InfluxData {
 	f.mux.RLock()
 	defer f.mux.RUnlock()
 	util.SetAttr(attr, in, metrics, f.lastMetadata[in.DeviceName])
-	ms := map[string]int64{}
-	msF := map[string]float64{}
-	for m, _ := range metrics {
+
+	profileName := "snmp"
+	results := []InfluxData{}
+	for m, name := range metrics {
+		if m == "" {
+			f.Errorf("Missing metric name, skipping %v", attr)
+			continue
+		}
+		profileName = name.Profile
 		if _, ok := in.CustomBigInt[m]; ok {
-			ms[m] = in.CustomBigInt[m]
+			attrNew := util.CopyAttrForSnmp(attr, m, name, f.lastMetadata[in.DeviceName])
+			if util.DropOnFilter(attrNew, f.lastMetadata[in.DeviceName], true) {
+				continue // This Metric isn't in the white list so lets drop it.
+			}
+
+			mtype := name.GetType()
+			if name.Format == kt.FloatMS {
+				results = append(results, InfluxData{
+					Name:        "kentik." + mtype,
+					FieldsFloat: map[string]float64{m: float64(float64(in.CustomBigInt[m]) / 1000)},
+					Timestamp:   in.Timestamp * 1000000000,
+					Tags:        attrNew,
+				})
+			} else {
+				results = append(results, InfluxData{
+					Name:      "kentik." + mtype,
+					Fields:    map[string]int64{m: int64(in.CustomBigInt[m])},
+					Timestamp: in.Timestamp * 1000000000,
+					Tags:      attrNew,
+				})
+			}
 		}
 	}
 
@@ -259,9 +305,19 @@ func (f *InfluxFormat) fromSnmpInterfaceMetric(in *kt.JCHF) []InfluxData {
 		if ii, ok := f.lastMetadata[in.DeviceName].InterfaceInfo[in.InputPort]; ok {
 			if speed, ok := ii["Speed"]; ok {
 				if ispeed, ok := speed.(int32); ok {
-					uptimeSpeed := in.CustomBigInt["Uptime"] * (int64(ispeed) * 1000000) // Convert into bits here, from megabits.
+					uptimeSpeed := in.CustomBigInt["Uptime"] * (int64(ispeed) * 10000) // Convert into bits here, from megabits. Also divide by 100 to convert uptime into seconds, from centi-seconds.
 					if uptimeSpeed > 0 {
-						msF["IfInUtilization"] = float64(in.CustomBigInt["ifHCInOctets"]*8*100) / float64(uptimeSpeed)
+						attrNew := util.CopyAttrForSnmp(attr, "IfInUtilization", kt.MetricInfo{Oid: "computed", Mib: "computed", Profile: profileName, Table: "if"}, f.lastMetadata[in.DeviceName])
+						if inBytes, ok := in.CustomBigInt["ifHCInOctets"]; ok {
+							if !util.DropOnFilter(attrNew, f.lastMetadata[in.DeviceName], true) {
+								results = append(results, InfluxData{
+									Name:        "kentik.snmp",
+									FieldsFloat: map[string]float64{"IfInUtilization": float64(inBytes*8*100) / float64(uptimeSpeed)},
+									Timestamp:   in.Timestamp * 1000000000,
+									Tags:        attrNew,
+								})
+							}
+						}
 					}
 				}
 			}
@@ -269,22 +325,26 @@ func (f *InfluxFormat) fromSnmpInterfaceMetric(in *kt.JCHF) []InfluxData {
 		if oi, ok := f.lastMetadata[in.DeviceName].InterfaceInfo[in.OutputPort]; ok {
 			if speed, ok := oi["Speed"]; ok {
 				if ispeed, ok := speed.(int32); ok {
-					uptimeSpeed := in.CustomBigInt["Uptime"] * (int64(ispeed) * 1000000) // Convert into bits here, from megabits.
+					uptimeSpeed := in.CustomBigInt["Uptime"] * (int64(ispeed) * 10000) // Convert into bits here, from megabits. Also divide by 100 to convert uptime into seconds, from centi-seconds.
 					if uptimeSpeed > 0 {
-						msF["IfOutUtilization"] = float64(in.CustomBigInt["ifHCOutOctets"]*8*100) / float64(uptimeSpeed)
+						attrNew := util.CopyAttrForSnmp(attr, "IfOutUtilization", kt.MetricInfo{Oid: "computed", Mib: "computed", Profile: profileName, Table: "if"}, f.lastMetadata[in.DeviceName])
+						if outBytes, ok := in.CustomBigInt["ifHCOutOctets"]; ok {
+							if !util.DropOnFilter(attrNew, f.lastMetadata[in.DeviceName], true) {
+								results = append(results, InfluxData{
+									Name:        "kentik.snmp",
+									FieldsFloat: map[string]float64{"IfOutUtilization": float64(outBytes*8*100) / float64(uptimeSpeed)},
+									Timestamp:   in.Timestamp * 1000000000,
+									Tags:        attrNew,
+								})
+							}
+						}
 					}
 				}
 			}
 		}
 	}
 
-	return []InfluxData{InfluxData{
-		Name:        "kentik.snmp.interface",
-		Fields:      ms,
-		FieldsFloat: msF,
-		Timestamp:   in.Timestamp * 1000000000,
-		Tags:        attr,
-	}}
+	return results
 }
 
 var escaper = regexp.MustCompile("([,= ])")
diff --git a/pkg/formats/nrm/nrm.go b/pkg/formats/nrm/nrm.go
index fde4395..3a2e23b 100644
--- a/pkg/formats/nrm/nrm.go
+++ b/pkg/formats/nrm/nrm.go
@@ -5,7 +5,6 @@ import (
 	"compress/gzip"
 	"fmt"
 	"os"
-	"sort"
 	"strconv"
 	"strings"
 	"sync"
@@ -34,8 +33,6 @@ const (
 	InstNameNetflowMetric = "netflow-metrics"
 	InstNameSynthetic     = "synthetic"
 	InstNameKtranslate    = "heartbeat"
-
-	MAX_ATTR_FOR_NR = 64
 )
 
 type NRMFormat struct {
@@ -477,7 +474,7 @@ func (f *NRMFormat) fromSnmpDeviceMetric(in *kt.JCHF) []NRMetric {
 			continue
 		}
 		if _, ok := in.CustomBigInt[m]; ok {
-			attrNew := copyAttrForSnmp(attr, m, name, f.lastMetadata[in.DeviceName])
+			attrNew := util.CopyAttrForSnmp(attr, m, name, f.lastMetadata[in.DeviceName])
 			if util.DropOnFilter(attrNew, f.lastMetadata[in.DeviceName], false) {
 				continue // This Metric isn't in the white list so lets drop it.
 			}
@@ -523,7 +520,7 @@ func (f *NRMFormat) fromSnmpInterfaceMetric(in *kt.JCHF) []NRMetric {
 		}
 		profileName = name.Profile
 		if _, ok := in.CustomBigInt[m]; ok {
-			attrNew := copyAttrForSnmp(attr, m, name, f.lastMetadata[in.DeviceName])
+			attrNew := util.CopyAttrForSnmp(attr, m, name, f.lastMetadata[in.DeviceName])
 			if util.DropOnFilter(attrNew, f.lastMetadata[in.DeviceName], true) {
 				continue // This Metric isn't in the white list so lets drop it.
 			}
@@ -555,7 +552,7 @@ func (f *NRMFormat) fromSnmpInterfaceMetric(in *kt.JCHF) []NRMetric {
 				if ispeed, ok := speed.(int32); ok {
 					uptimeSpeed := in.CustomBigInt["Uptime"] * (int64(ispeed) * 10000) // Convert into bits here, from megabits. Also divide by 100 to convert uptime into seconds, from centi-seconds.
 					if uptimeSpeed > 0 {
-						attrNew := copyAttrForSnmp(attr, "IfInUtilization", kt.MetricInfo{Oid: "computed", Mib: "computed", Profile: profileName, Table: "if"}, f.lastMetadata[in.DeviceName])
+						attrNew := util.CopyAttrForSnmp(attr, "IfInUtilization", kt.MetricInfo{Oid: "computed", Mib: "computed", Profile: profileName, Table: "if"}, f.lastMetadata[in.DeviceName])
 						if inBytes, ok := in.CustomBigInt["ifHCInOctets"]; ok {
 							if !util.DropOnFilter(attrNew, f.lastMetadata[in.DeviceName], true) {
 								ms = append(ms, NRMetric{
@@ -575,7 +572,7 @@ func (f *NRMFormat) fromSnmpInterfaceMetric(in *kt.JCHF) []NRMetric {
 				if ispeed, ok := speed.(int32); ok {
 					uptimeSpeed := in.CustomBigInt["Uptime"] * (int64(ispeed) * 10000) // Convert into bits here, from megabits. Also divide by 100 to convert uptime into seconds, from centi-seconds.
 					if uptimeSpeed > 0 {
-						attrNew := copyAttrForSnmp(attr, "IfOutUtilization", kt.MetricInfo{Oid: "computed", Mib: "computed", Profile: profileName, Table: "if"}, f.lastMetadata[in.DeviceName])
+						attrNew := util.CopyAttrForSnmp(attr, "IfOutUtilization", kt.MetricInfo{Oid: "computed", Mib: "computed", Profile: profileName, Table: "if"}, f.lastMetadata[in.DeviceName])
 						if outBytes, ok := in.CustomBigInt["ifHCOutOctets"]; ok {
 							if !util.DropOnFilter(attrNew, f.lastMetadata[in.DeviceName], true) {
 								ms = append(ms, NRMetric{
@@ -671,123 +668,6 @@ func (f *NRMFormat) fromKtranslate(in *kt.JCHF) []NRMetric {
 	return ms
 }
 
-// List of attributes to not pass to NR.
-var removeAttrForSnmp = []string{
-	"Uptime",
-	"if_LastChange",
-	"SysServices",
-	"if_Mtu",
-	"if_ConnectorPresent",
-	"output_port",
-	"input_port",
-	"DropMetric",
-	"sysoid_vendor",
-}
-
-var keepAcrossTables = map[string]bool{
-	"device_name":    true,
-	"eventType":      true,
-	"provider":       true,
-	"sysoid_profile": true,
-	kt.IndexVar:      true,
-	"if_Index":       true,
-}
-
-var allowSysAttr = map[string]bool{
-	"Uptime":   true,
-	"MinRttMs": true,
-	"MaxRttMs": true,
-	"AvgRttMs": true,
-}
-
-func copyAttrForSnmp(attr map[string]interface{}, metricName string, name kt.MetricInfo, lm *kt.LastMetadata) map[string]interface{} {
-	attrNew := map[string]interface{}{
-		"objectIdentifier":     name.Oid,
-		"mib-name":             name.Mib,
-		"instrumentation.name": name.Profile,
-	}
-
-	// If set, add this in.
-	durSec := name.PollDur.Seconds()
-	if durSec > 0 {
-		attrNew["poll_duration_sec"] = name.PollDur.Seconds() + kt.PollAdjustTime
-	}
-
-	for k, v := range attr {
-		if !allowSysAttr[metricName] { // Only allow Sys* attributes on specific metrics.
-			if strings.HasPrefix(k, "Sys") || k == "src_addr" {
-				continue
-			}
-		}
-
-		newKey := k
-		if strings.HasPrefix(k, kt.StringPrefix) {
-			newKey = k[len(kt.StringPrefix):]
-		}
-
-		if name.Table != "" && metricName != newKey {
-			if _, ok := keepAcrossTables[newKey]; !ok { // If we want this attribute in every table, list it here.
-				attrNew["mib-table"] = name.Table
-
-				// See if the metadata knows about this attribute.
-				if tableName, allNames, ok := lm.GetTableName(newKey); ok && len(allNames) > 0 {
-					if !allNames[name.Table] && tableName != kt.DeviceTagTable {
-						continue
-					}
-				} else {
-					// If this metric comes from a specific table, only show attributes for this table.
-					if !strings.HasPrefix(newKey, name.Table) {
-						if !allNames[name.Table] && tableName != kt.DeviceTagTable {
-							continue
-						}
-					}
-				}
-			}
-		}
-
-		// Case where metric has no table.
-		if name.Table == "" {
-			if tableName, _, ok := lm.GetTableName(newKey); ok {
-				if tableName != "" && tableName != kt.DeviceTagTable {
-					continue
-				}
-			}
-		}
-
-		attrNew[newKey] = v
-	}
-
-	// Delete a few attributes we don't want adding to cardinality.
-	for _, key := range removeAttrForSnmp {
-		delete(attrNew, key)
-	}
-
-	if len(attrNew) > MAX_ATTR_FOR_NR {
-		// Since NR limits us to 100 attributes, we need to prune. Take the first 100 lexographical keys.
-		keys := make([]string, len(attrNew))
-		i := 0
-		for k, _ := range attrNew {
-			keys[i] = k
-			i++
-		}
-		sort.Strings(keys)
-		for _, k := range keys[MAX_ATTR_FOR_NR-3:] {
-			delete(attrNew, k)
-		}
-
-		// Force these to be back in.
-		attrNew["objectIdentifier"] = name.Oid
-		attrNew["mib-name"] = name.Mib
-		attrNew["instrumentation.name"] = name.Profile
-	}
-
-	if attrNew["mib-name"] == "" {
-		delete(attrNew, "mib-name")
-	}
-
-	return attrNew
-}
-
 func toInstName(prov kt.Provider) string {
 	if strings.Contains(string(prov), "vpc") {
 		return InstNameVPCMetric
diff --git a/pkg/formats/util/util.go b/pkg/formats/util/util.go
index c29e790..f42c1d6 100644
--- a/pkg/formats/util/util.go
+++ b/pkg/formats/util/util.go
@@ -2,6 +2,7 @@ package util
 
 import (
 	"fmt"
+	"sort"
 	"strconv"
 	"strings"
 
@@ -34,6 +35,7 @@ var (
 
 const (
 	FORCE_MATCH_TOKEN = "!"
+	MAX_ATTR_FOR_SNMP = 64
 )
 
 func SetAttr(attr map[string]interface{}, in *kt.JCHF, metrics map[string]kt.MetricInfo, lastMetadata *kt.LastMetadata) {
@@ -259,3 +261,120 @@ var (
 func GetSynMetricNameSet(rt int32) map[string]kt.MetricInfo {
 	return synMetrics[rt]
 }
+
+// List of attributes to not pass to NR.
+var removeAttrForSnmp = []string{
+	"Uptime",
+	"if_LastChange",
+	"SysServices",
+	"if_Mtu",
+	"if_ConnectorPresent",
+	"output_port",
+	"input_port",
+	"DropMetric",
+	"sysoid_vendor",
+}
+
+var keepAcrossTables = map[string]bool{
+	"device_name":    true,
+	"eventType":      true,
+	"provider":       true,
+	"sysoid_profile": true,
+	kt.IndexVar:      true,
+	"if_Index":       true,
+}
+
+var allowSysAttr = map[string]bool{
+	"Uptime":   true,
+	"MinRttMs": true,
+	"MaxRttMs": true,
+	"AvgRttMs": true,
+}
+
+func CopyAttrForSnmp(attr map[string]interface{}, metricName string, name kt.MetricInfo, lm *kt.LastMetadata) map[string]interface{} {
+	attrNew := map[string]interface{}{
+		"objectIdentifier":     name.Oid,
+		"mib-name":             name.Mib,
+		"instrumentation.name": name.Profile,
+	}
+
+	// If set, add this in.
+	durSec := name.PollDur.Seconds()
+	if durSec > 0 {
+		attrNew["poll_duration_sec"] = name.PollDur.Seconds() + kt.PollAdjustTime
+	}
+
+	for k, v := range attr {
+		if !allowSysAttr[metricName] { // Only allow Sys* attributes on specific metrics.
+			if strings.HasPrefix(k, "Sys") || k == "src_addr" {
+				continue
+			}
+		}
+
+		newKey := k
+		if strings.HasPrefix(k, kt.StringPrefix) {
+			newKey = k[len(kt.StringPrefix):]
+		}
+
+		if name.Table != "" && metricName != newKey {
+			if _, ok := keepAcrossTables[newKey]; !ok { // If we want this attribute in every table, list it here.
+				attrNew["mib-table"] = name.Table
+
+				// See if the metadata knows about this attribute.
+				if tableName, allNames, ok := lm.GetTableName(newKey); ok && len(allNames) > 0 {
+					if !allNames[name.Table] && tableName != kt.DeviceTagTable {
+						continue
+					}
+				} else {
+					// If this metric comes from a specific table, only show attributes for this table.
+					if !strings.HasPrefix(newKey, name.Table) {
+						if !allNames[name.Table] && tableName != kt.DeviceTagTable {
+							continue
+						}
+					}
+				}
+			}
+		}
+
+		// Case where metric has no table.
+		if name.Table == "" {
+			if tableName, _, ok := lm.GetTableName(newKey); ok {
+				if tableName != "" && tableName != kt.DeviceTagTable {
+					continue
+				}
+			}
+		}
+
+		attrNew[newKey] = v
+	}
+
+	// Delete a few attributes we don't want adding to cardinality.
+	for _, key := range removeAttrForSnmp {
+		delete(attrNew, key)
+	}
+
+	if len(attrNew) > MAX_ATTR_FOR_SNMP {
+		// Since NR limits us to 100 attributes, we need to prune. Take the first 100 lexographical keys.
+		keys := make([]string, len(attrNew))
+		i := 0
+		for k, _ := range attrNew {
+			keys[i] = k
+			i++
+		}
+		sort.Strings(keys)
+		for _, k := range keys[MAX_ATTR_FOR_SNMP-3:] {
+			delete(attrNew, k)
+		}
+
+		// Force these to be back in.
+		attrNew["objectIdentifier"] = name.Oid
+		attrNew["mib-name"] = name.Mib
+		attrNew["instrumentation.name"] = name.Profile
+	}
+
+	if attrNew["mib-name"] == "" {
+		delete(attrNew, "mib-name")
+	}
+
+	return attrNew
+}
diff --git a/pkg/inputs/snmp/disco.go b/pkg/inputs/snmp/disco.go
index 8f05814..a4ce8af 100644
--- a/pkg/inputs/snmp/disco.go
+++ b/pkg/inputs/snmp/disco.go
@@ -23,7 +23,8 @@ import (
 )
 
 const (
-	permissions = 0644
+	permissions      = 0644
+	RemoveTimeBuffer = -1 * 15 * 60 * time.Second // This is a 15 min buffer time on removing a device which isn't seen.
 )
 
 type SnmpDiscoDeviceStat struct {
@@ -32,7 +33,7 @@ type SnmpDiscoDeviceStat struct {
 	delta    int
 }
 
-func Discover(ctx context.Context, snmpFile string, log logger.ContextL) (*SnmpDiscoDeviceStat, error) {
+func Discover(ctx context.Context, snmpFile string, log logger.ContextL, pollDuration time.Duration) (*SnmpDiscoDeviceStat, error) {
 	// First, parse the config file and see what we're doing.
 	log.Infof("SNMP Discovery, loading config from %s", snmpFile)
 	conf, err := parseConfig(ctx, snmpFile, log)
@@ -54,7 +55,7 @@ func Discover(ctx context.Context, snmpFile string, log logger.ContextL) (*SnmpD
 	}
 
 	if conf.Disco.AddDevices { // Verify that the output is writeable before diving into discoing.
-		if _, err := addDevices(ctx, nil, snmpFile, conf, true, log); err != nil {
+		if _, err := addDevices(ctx, nil, snmpFile, conf, true, log, pollDuration); err != nil {
 			return nil, fmt.Errorf("There was an error when writing the %s SNMP configuration file: %v.", snmpFile, err)
 		}
 	}
@@ -128,7 +129,7 @@ func Discover(ctx context.Context, snmpFile string, log logger.ContextL) (*SnmpD
 
 	var stats *SnmpDiscoDeviceStat
 	if conf.Disco.AddDevices {
-		stats, err = addDevices(ctx, foundDevices, snmpFile, conf, false, log)
+		stats, err = addDevices(ctx, foundDevices, snmpFile, conf, false, log, pollDuration)
 		if err != nil {
 			return nil, err
 		}
@@ -146,7 +147,7 @@ func RunDiscoOnTimer(ctx context.Context, c chan os.Signal, snmpFile string, log
 	defer discoCheck.Stop()
 
 	check := func() {
-		stats, err := Discover(ctx, snmpFile, log)
+		stats, err := Discover(ctx, snmpFile, log, pt)
 		if err != nil {
 			log.Errorf("Discovery SNMP Error: %v", err)
 		} else {
@@ -302,7 +303,7 @@ func doubleCheckHost(result scan.Result, timeout time.Duration, ctl chan bool, m
 	foundDevices[result.Host.String()] = &device
 }
 
-func addDevices(ctx context.Context, foundDevices map[string]*kt.SnmpDeviceConfig, snmpFile string, conf *kt.SnmpConfig, isTest bool, log logger.ContextL) (*SnmpDiscoDeviceStat, error) {
+func addDevices(ctx context.Context, foundDevices map[string]*kt.SnmpDeviceConfig, snmpFile string, conf *kt.SnmpConfig, isTest bool, log logger.ContextL, pollDuration time.Duration) (*SnmpDiscoDeviceStat, error) {
 	// Now add the new.
 	stats := SnmpDiscoDeviceStat{}
 	if conf.Devices == nil {
@@ -360,6 +361,34 @@ func addDevices(ctx context.Context, foundDevices map[string]*kt.SnmpDeviceConfi
 		}
 	}
 
+	// Remove any devices which havn't been up long enough if this is set.
+	if !isTest {
+		removeNum := conf.Global.PurgeDevices
+		for dip, d := range conf.Devices {
+			if d.PurgeDevice != 0 { // Let this override the global settings.
+				removeNum = d.PurgeDevice
+			}
+
+			if removeNum == -1 { // If this value is -1, keep this device forever.
+				continue
+			} else if removeNum < -1 {
+				// This is an invalid state, throw an error.
+				return nil, fmt.Errorf("Invalid PurgeDevice value (%d) for device %s", removeNum, d.DeviceName)
+			}
+
+			if removeNum > 0 && pollDuration > 0 { // Keep remove num guard just in case.
+				// Get time this would take.
+				removeTime := time.Now().Add(time.Duration(removeNum) * pollDuration * -1)
+				removeTime = removeTime.Add(RemoveTimeBuffer)
+				if d.Checked.Before(removeTime) {
+					log.Infof("Removing device %s because it hasn't been seen since %v. Max time %v", d.DeviceName, d.Checked, removeTime)
+					delete(conf.Devices, dip)
+					stats.added--
+				}
+			}
+		}
+	}
+
 	// Calculate total number of new devices.
 	stats.delta = origCount - len(conf.Devices)
 	if !isTest {
diff --git a/pkg/inputs/snmp/metrics/device_metrics.go b/pkg/inputs/snmp/metrics/device_metrics.go
index 965575e..d620857 100644
--- a/pkg/inputs/snmp/metrics/device_metrics.go
+++ b/pkg/inputs/snmp/metrics/device_metrics.go
@@ -161,6 +161,11 @@ func (dm *DeviceMetrics) pollFromConfig(ctx context.Context, server *gosnmp.GoSN
 		}
 		oidName := wrapper.mib.GetName()
 
+		// This result is blocked due to a defined condition.
+		if !wrapper.checkCondition(idx, results) {
+			continue
+		}
+
 		dmr := assureDeviceMetrics(m, idx)
 		metricsFound[oidName] = kt.MetricInfo{Oid: wrapper.mib.Oid, Mib: wrapper.mib.Mib, Profile: dm.profileName, Table: wrapper.mib.Table, PollDur: wrapper.mib.PollDur}
 		switch wrapper.variable.Type {
@@ -377,3 +382,21 @@ func (dm *DeviceMetrics) GetPingStats(ctx context.Context, pinger *ping.Pinger)
 
 	return []*kt.JCHF{dst}, nil
 }
+
+func (w wrapper) checkCondition(idx string, results []wrapper) bool { // Check condition, if it exists. If this is false, we skip this result.
+	if w.mib.Condition != nil {
+		for _, wr := range results { // Annoying we have to itterate twice here.
+			idxw := snmp_util.GetIndex(wr.variable.Name[1:], wr.oid)
+			if idxw != idx { // We only care about results which share a common index.
+				continue
+			}
+			oidNameWr := wr.mib.GetName() // Does the name match our target?
+			if oidNameWr == w.mib.Condition.TargetName {
+				// If it does, does it match our target value?
+				return snmp_util.ToInt64(wr.variable.Value) == w.mib.Condition.TargetValue
+			}
+		}
+		return true // Keep this one because condition evals to true.
+	}
+	return true // True if no condition exists.
+}
diff --git a/pkg/inputs/snmp/mibs/profile.go b/pkg/inputs/snmp/mibs/profile.go
index d9acf1c..46a426f 100644
--- a/pkg/inputs/snmp/mibs/profile.go
+++ b/pkg/inputs/snmp/mibs/profile.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"os"
 	"regexp"
+	"strconv"
 	"strings"
 	"time"
 
@@ -27,6 +28,7 @@ type OID struct {
 	MatchAttr  []string         `yaml:"match_attributes,omitempty"`
 	Format     string           `yaml:"format,omitempty"`
 	AllowDup   bool             `yaml:"allow_duplicate,omitempty"`
+	Condition  string           `yaml:"condition,omitempty"`
 }
 
 type Tag struct {
@@ -436,6 +438,7 @@ func (p *Profile) GetMetrics(enabledMibs []string, counterTimeSec int) (map[stri
 				FromExtended: metric.fromExtended,
 				Format:       metric.Symbol.Format,
 				AllowDup:     metric.Symbol.AllowDup,
+				Condition:    metric.Symbol.GetCondition(p),
 			}
 			if len(mib.Enum) > 0 {
 				mib.EnumRev = make(map[int64]string)
@@ -458,6 +461,9 @@ func (p *Profile) GetMetrics(enabledMibs []string, counterTimeSec int) (map[stri
 				p.Warnf("Skipping mib with no name: %v", mib)
 				continue
 			}
+			if mib.Condition != nil {
+				p.Infof("SNMP: Condition of %s=%d for %s", mib.Condition.TargetName, mib.Condition.TargetValue, mib.Name)
+			}
 			if metric.IsInterface || strings.HasPrefix(metric.Symbol.Name, "if") {
 				interfaceMetrics[metric.Symbol.Oid] = mib
 			} else {
@@ -478,6 +484,7 @@ func (p *Profile) GetMetrics(enabledMibs []string, counterTimeSec int) (map[stri
 				FromExtended: metric.fromExtended,
 				Format:       s.Format,
 				AllowDup:     s.AllowDup,
+				Condition:    s.GetCondition(p),
 			}
 			if len(mib.Enum) > 0 {
 				mib.EnumRev = make(map[int64]string)
@@ -500,6 +507,9 @@ func (p *Profile) GetMetrics(enabledMibs []string, counterTimeSec int) (map[stri
 				p.Warnf("Skipping mib with no name: %v", mib)
 				continue
 			}
+			if mib.Condition != nil {
+				p.Infof("SNMP: Condition of %s=%d for %s", mib.Condition.TargetName, mib.Condition.TargetValue, mib.Name)
+			}
 			if metric.IsInterface || strings.HasPrefix(s.Name, "if") {
 				interfaceMetrics[s.Oid] = mib
 			} else {
@@ -858,3 +868,23 @@ func (o *OID) GetTableName() string {
 	}
 	return o.Name
 }
+
+func (o *OID) GetCondition(log logger.ContextL) *kt.MibCondition {
+	if o.Condition != "" {
+		pts := strings.Split(o.Condition, "=")
+		if len(pts) == 2 {
+			val, err := strconv.Atoi(pts[1])
+			if err != nil {
+				log.Errorf("Skipping invalid profile condition in %s: %s. RHS (%s) must be an int and operator must be '='.", o.Name, o.Condition, pts[1])
+			} else {
+				return &kt.MibCondition{
+					TargetName:  pts[0],
+					TargetValue: int64(val),
+				}
+			}
+		} else {
+			log.Errorf("Skipping invalid profile condition in %s: %s. RHS must be an int and operator must be '='.", o.Name, o.Condition)
+		}
+	}
+	return nil
+}
diff --git a/pkg/kt/snmp.go b/pkg/kt/snmp.go
index 77fd557..dfb468a 100644
--- a/pkg/kt/snmp.go
+++ b/pkg/kt/snmp.go
@@ -186,6 +186,7 @@ type SnmpDeviceConfig struct {
 	RunPing             bool              `yaml:"response_time,omitempty"`
 	Ext                 *ExtensionSet     `yaml:"ext,omitempty"`
 	PingSec             int               `yaml:"ping_interval_sec,omitempty"`
+	PurgeDevice         int               `yaml:"purge_after_num"` // Delete this device if its not seen after X discovery attempts. Default is 0, which means things never get purged unless global value is true. Set to -1 to pin device always.
 	allUserTags         map[string]string
 	walker              SNMPTestWalker
 }
@@ -228,6 +229,7 @@ type SnmpGlobalConfig struct {
 	GlobalV3      *V3SNMPConfig     `yaml:"global_v3"`
 	RunPing       bool              `yaml:"response_time"`
 	PingSec       int               `yaml:"ping_interval_sec,omitempty"`
+	PurgeDevices  int               `yaml:"purge_devices_after_num"` // Delete any device if its not seen after X discovery attempts. Default is 0, which means things never get purged.
 	UserTags      map[string]string `yaml:"user_tags"`
 	MatchAttr     map[string]string `yaml:"match_attributes"`
 }
@@ -318,6 +320,11 @@ const (
 	Integer32         = 16
 )
 
+type MibCondition struct {
+	TargetName  string
+	TargetValue int64
+}
+
 type Mib struct {
 	Oid          string
 	Name         string
@@ -336,6 +343,7 @@ type Mib struct {
 	OtherTables  map[string]bool
 	Format       string
 	AllowDup     bool
+	Condition    *MibCondition
 }
 
 func (mb *Mib) String() string {
@@ -572,6 +580,9 @@ func (d *SnmpDeviceConfig) UpdateFrom(old *SnmpDeviceConfig, conf *SnmpConfig) {
 	if old.RunPing {
 		d.RunPing = old.RunPing
 	}
+	if old.PurgeDevice != 0 {
+		d.PurgeDevice = old.PurgeDevice
+	}
 	if d.UserTags == nil && old.UserTags != nil {
 		d.UserTags = map[string]string{}
 	}


Test patch:
diff --git a/pkg/formats/nrm/nrm_test.go b/pkg/formats/nrm/nrm_test.go
deleted file mode 100644
index 249b6e3..0000000
--- a/pkg/formats/nrm/nrm_test.go
+++ /dev/null
@@ -1,54 +0,0 @@
-package nrm
-
-import (
-	"fmt"
-	"testing"
-
-	"github.com/kentik/ktranslate/pkg/kt"
-	"github.com/stretchr/testify/assert"
-)
-
-func TestCopyAttrforSNMP(t *testing.T) {
-	assert := assert.New(t)
-
-	input := map[string]interface{}{}
-	for i := 0; i < 10; i++ {
-		input[fmt.Sprintf("XXX%d", i)] = i
-	}
-	name := kt.MetricInfo{Oid: "oid", Mib: "mib"}
-
-	res := copyAttrForSnmp(input, "test", name, nil)
-	assert.Equal(len(input)+3, len(res)) // adds in three keys
-	assert.Equal("oid", res["objectIdentifier"])
-
-	for i := 0; i < MAX_ATTR_FOR_NR+10; i++ {
-		input[fmt.Sprintf("XXX%d", i)] = i
-	}
-	res = copyAttrForSnmp(input, "test", name, nil)
-	assert.Equal(MAX_ATTR_FOR_NR, len(res)) // truncated at MAX_ATTR_FOR_NR
-	assert.Equal("oid", res["objectIdentifier"])
-
-	input = map[string]interface{}{kt.StringPrefix + "foo": "one"}
-	res = copyAttrForSnmp(input, "test", name, nil)
-	assert.Equal("one", res["foo"], res)
-
-	input = map[string]interface{}{kt.StringPrefix + "foo": "one"}
-	name = kt.MetricInfo{Oid: "oid", Mib: "mib", Table: "noMatch"}
-	res = copyAttrForSnmp(input, "test", name, nil)
-	assert.Equal(nil, res["foo"], res)
-
-	input = map[string]interface{}{kt.StringPrefix + "foo": "one"}
-	name = kt.MetricInfo{Oid: "oid", Mib: "mib", Table: "foo"}
-	res = copyAttrForSnmp(input, "test", name, nil)
-	assert.Equal("one", res["foo"], res)
-
-	input = map[string]interface{}{"foo": "one"}
-	name = kt.MetricInfo{Oid: "oid", Mib: "mib", Table: "foo"}
-	res = copyAttrForSnmp(input, "test", name, nil)
-	assert.Equal("one", res["foo"], res)
-
-	input = map[string]interface{}{"foo": "one"}
-	name = kt.MetricInfo{Oid: "oid", Mib: "mib", Table: "somethingElse"}
-	res = copyAttrForSnmp(input, "test", name, nil)
-	assert.Equal(nil, res["foo"], res)
-}
diff --git a/pkg/formats/util/util_test.go b/pkg/formats/util/util_test.go
index 8b04e31..9a3d3b8 100644
--- a/pkg/formats/util/util_test.go
+++ b/pkg/formats/util/util_test.go
@@ -1,6 +1,7 @@
 package util
 
 import (
+	"fmt"
 	"regexp"
 	"strings"
 	"testing"
@@ -258,3 +259,48 @@ func TestDropOnFilter(t *testing.T) {
 		assert.Equal(t, test.drop, drop, "Test %d", i)
 	}
 }
+
+func TestCopyAttrforSNMP(t *testing.T) {
+	assert := assert.New(t)
+
+	input := map[string]interface{}{}
+	for i := 0; i < 10; i++ {
+		input[fmt.Sprintf("XXX%d", i)] = i
+	}
+	name := kt.MetricInfo{Oid: "oid", Mib: "mib"}
+
+	res := CopyAttrForSnmp(input, "test", name, nil)
+	assert.Equal(len(input)+3, len(res)) // adds in three keys
+	assert.Equal("oid", res["objectIdentifier"])
+
+	for i := 0; i < MAX_ATTR_FOR_SNMP+10; i++ {
+		input[fmt.Sprintf("XXX%d", i)] = i
+	}
+	res = CopyAttrForSnmp(input, "test", name, nil)
+	assert.Equal(MAX_ATTR_FOR_SNMP, len(res)) // truncated at MAX_ATTR_FOR_SNMP
+	assert.Equal("oid", res["objectIdentifier"])
+
+	input = map[string]interface{}{kt.StringPrefix + "foo": "one"}
+	res = CopyAttrForSnmp(input, "test", name, nil)
+	assert.Equal("one", res["foo"], res)
+
+	input = map[string]interface{}{kt.StringPrefix + "foo": "one"}
+	name = kt.MetricInfo{Oid: "oid", Mib: "mib", Table: "noMatch"}
+	res = CopyAttrForSnmp(input, "test", name, nil)
+	assert.Equal(nil, res["foo"], res)
+
+	input = map[string]interface{}{kt.StringPrefix + "foo": "one"}
+	name = kt.MetricInfo{Oid: "oid", Mib: "mib", Table: "foo"}
+	res = CopyAttrForSnmp(input, "test", name, nil)
+	assert.Equal("one", res["foo"], res)
+
+	input = map[string]interface{}{"foo": "one"}
+	name = kt.MetricInfo{Oid: "oid", Mib: "mib", Table: "foo"}
+	res = CopyAttrForSnmp(input, "test", name, nil)
+	assert.Equal("one", res["foo"], res)
+
+	input = map[string]interface{}{"foo": "one"}
+	name = kt.MetricInfo{Oid: "oid", Mib: "mib", Table: "somethingElse"}
+	res = CopyAttrForSnmp(input, "test", name, nil)
+	assert.Equal(nil, res["foo"], res)
+}
diff --git a/pkg/inputs/snmp/metrics/device_metrics_test.go b/pkg/inputs/snmp/metrics/device_metrics_test.go
new file mode 100644
index 0000000..2a316f5
--- /dev/null
+++ b/pkg/inputs/snmp/metrics/device_metrics_test.go
@@ -0,0 +1,50 @@
+package metrics
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/gosnmp/gosnmp"
+	"github.com/kentik/ktranslate/pkg/inputs/snmp/mibs"
+	"github.com/kentik/ktranslate/pkg/kt"
+)
+
+func TestCheckCondition(t *testing.T) {
+	assert := assert.New(t)
+
+	tests := map[string]bool{
+		"":         true,
+		"name1=2":  false,
+		"name1=66": true,
+	}
+
+	resultSet := []wrapper{
+		wrapper{
+			oid:      "1.2",
+			mib:      &kt.Mib{Tag: "name1"},
+			variable: gosnmp.SnmpPDU{Value: 66, Name: "1.2.5"},
+		},
+		wrapper{
+			oid:      "1.1",
+			mib:      &kt.Mib{Tag: "name2"},
+			variable: gosnmp.SnmpPDU{Value: 3, Name: "1.1.5"},
+		},
+	}
+
+	for in, expt := range tests {
+		oid := mibs.OID{Condition: in}
+		mib := &kt.Mib{
+			Condition: oid.GetCondition(nil),
+			Tag:       "name2",
+		}
+		w := wrapper{
+			variable: gosnmp.SnmpPDU{Value: 3, Name: "1.1.5"},
+			mib:      mib,
+			oid:      "1.1",
+		}
+
+		res := w.checkCondition(".5", resultSet)
+		assert.Equal(expt, res, "%s", in)
+	}
+}