Repository: kentik/ktranslate
Pull request #204 (2021-11-23 21:18:55)

Uptime Fixes
Moving interface metrics to count type, from gauge. 

Also fixing utilization algorithm.


Patch:
diff --git a/cmd/ktranslate/main.go b/cmd/ktranslate/main.go
index f11565e..c01f146 100644
--- a/cmd/ktranslate/main.go
+++ b/cmd/ktranslate/main.go
@@ -33,8 +33,8 @@ func main() {
 		threads        = flag.Int("threads", 0, "Number of threads to run for processing")
 		threadsInput   = flag.Int("input_threads", 0, "Number of threads to run for input processing")
 		maxThreads     = flag.Int("max_threads", 0, "Dynamically grow threads up to this number")
-		format         = flag.String("format", "flat_json", "Format to convert kflow to: (json|flat_json|avro|netflow|influx|prometheus|new_relic|new_relic_metric|splunk)")
-		formatRollup   = flag.String("format_rollup", "", "Format to convert rollups to: (json|avro|netflow|influx|prometheus|new_relic|new_relic_metric|splunk)")
+		format         = flag.String("format", "flat_json", "Format to convert kflow to: (json|flat_json|avro|netflow|influx|prometheus|new_relic|new_relic_metric|splunk|elasticsearch)")
+		formatRollup   = flag.String("format_rollup", "", "Format to convert rollups to: (json|avro|netflow|influx|prometheus|new_relic|new_relic_metric|splunk|elasticsearch)")
 		compression    = flag.String("compression", "none", "compression algo to use (none|gzip|snappy|deflate|null)")
 		sinks          = flag.String("sinks", "stdout", "List of sinks to send data to. Options: (kafka|stdout|new_relic|kentik|net|http|splunk|prometheus|file|s3|gcloud)")
 		maxFlows       = flag.Int("max_flows_per_message", 10000, "Max number of flows to put in each emitted message")
diff --git a/pkg/formats/elasticsearch/elasticsearch.go b/pkg/formats/elasticsearch/elasticsearch.go
new file mode 100644
index 0000000..6adba7b
--- /dev/null
+++ b/pkg/formats/elasticsearch/elasticsearch.go
@@ -0,0 +1,186 @@
+package elasticsearch
+
+import (
+	"bufio"
+	"bytes"
+	"compress/gzip"
+	"fmt"
+	"io"
+	"strings"
+
+	"github.com/kentik/ktranslate/pkg/formats/util"
+	"github.com/kentik/ktranslate/pkg/kt"
+	"github.com/kentik/ktranslate/pkg/rollup"
+
+	jsoniter "github.com/json-iterator/go"
+	"github.com/kentik/ktranslate/pkg/eggs/logger"
+)
+
+const (
+	InstNameNetflowEvent = "netflow-events"
+	InstNameVPCEvent     = "vpc-flow-events"
+	InstNameAWSVPCEvent  = "aws-vpc-flow-events"
+	actionEntry          = `{"index":{}}`
+)
+
+var json = jsoniter.ConfigFastest
+
+type ElasticsearchFormat struct {
+	logger.ContextL
+	compression kt.Compression
+	useGzip     bool
+}
+
+func NewFormat(log logger.Underlying, compression kt.Compression) (*ElasticsearchFormat, error) {
+	ef := &ElasticsearchFormat{
+		ContextL:    logger.NewContextLFromUnderlying(logger.SContext{S: "elasticsearchFormat"}, log),
+		compression: compression,
+	}
+
+	switch compression {
+	case kt.CompressionNone:
+		ef.useGzip = false
+	case kt.CompressionGzip:
+		ef.useGzip = true
+	default:
+		return nil, fmt.Errorf("Invalid compression (%s): format json only supports none|gzip", compression)
+	}
+
+	return ef, nil
+}
+
+func (f *ElasticsearchFormat) To(msgs []*kt.JCHF, serBuf []byte) (*kt.Output, error) {
+	msgsNew := make([]map[string]interface{}, 0, len(msgs))
+	for _, msg := range msgs {
+		if msg.EventType == kt.KENTIK_EVENT_SNMP {
+			continue
+		}
+		mm := msg.Flatten()
+		strip(mm)
+		msgsNew = append(msgsNew, mm)
+	}
+
+	esBulkData := []string{}
+	for _, m := range msgsNew {
+		data, err := serialize(m)
+		if err != nil {
+			return nil, err
+		}
+		esBulkData = append(esBulkData, data)
+	}
+
+	esData := []byte(strings.Join(esBulkData, ""))
+
+	// compression
+	if f.useGzip {
+		buf := bytes.NewBuffer(serBuf)
+		buf.Reset()
+		zw, err := gzip.NewWriterLevel(buf, gzip.DefaultCompression)
+		if err != nil {
+			return nil, err
+		}
+
+		if _, err := zw.Write(esData); err != nil {
+			return nil, err
+		}
+
+		if err := zw.Close(); err != nil {
+			return nil, err
+		}
+		esData = buf.Bytes()
+	}
+	return kt.NewOutputWithProvider(esData, msgs[0].Provider, kt.EventOutput), nil
+}
+
+func (f *ElasticsearchFormat) From(raw *kt.Output) ([]map[string]interface{}, error) {
+	msgs := []*kt.JCHF{}
+
+	var r io.Reader
+	r = bytes.NewReader(raw.Body)
+	if f.useGzip {
+		gr, err := gzip.NewReader(bytes.NewBuffer(raw.Body))
+		if err != nil {
+			return nil, err
+		}
+		r = gr
+	}
+	sc := bufio.NewScanner(r)
+	for sc.Scan() {
+		// check for ES action and ignore
+		v := sc.Text()
+		if v == actionEntry {
+			continue
+		}
+		msg := &kt.JCHF{}
+		if err := json.Unmarshal([]byte(v), &msg); err != nil {
+			return nil, err
+		}
+		msgs = append(msgs, msg)
+	}
+
+	values := make([]map[string]interface{}, len(msgs))
+	for i, m := range msgs {
+		m.SetMap()
+		values[i] = m.ToMap()
+	}
+
+	return values, nil
+}
+
+func (f *ElasticsearchFormat) Rollup(rolls []rollup.Rollup) (*kt.Output, error) {
+	// serialize rolls
+	esBulkData := []string{}
+	for _, m := range rolls {
+		data, err := serialize(m)
+		if err != nil {
+			return nil, err
+		}
+		esBulkData = append(esBulkData, data)
+	}
+	return kt.NewOutputWithProvider([]byte(strings.Join(esBulkData, "")), rolls[0].Provider, kt.RollupOutput), nil
+}
+
+func serialize(o interface{}) (string, error) {
+	s := actionEntry + "\n"
+	data, err := json.Marshal(o)
+	if err != nil {
+		return "", err
+	}
+	s += string(data) + "\n"
+	return s, nil
+}
+
+func strip(in map[string]interface{}) {
+	for k, v := range in {
+		switch tv := v.(type) {
+		case string:
+			if tv == "" || tv == "-" || tv == "--" {
+				delete(in, k)
+			}
+		case int32:
+			if tv == 0 {
+				delete(in, k)
+			}
+		case int64:
+			if tv == 0 {
+				delete(in, k)
+			}
+		}
+		if _, ok := util.DroppedAttrs[k]; ok {
+			delete(in, k)
+		}
+	}
+	in["instrumentation.provider"] = kt.InstProvider // Let them know who sent this.
+	switch in["provider"] {
+	case kt.ProviderVPC:
+		switch in["kt.from"] {
+		case kt.FromLambda:
+			in["instrumentation.name"] = InstNameAWSVPCEvent
+		default:
+			in["instrumentation.name"] = InstNameVPCEvent
+		}
+	default:
+		in["instrumentation.name"] = InstNameNetflowEvent
+	}
+	in["collector.name"] = kt.CollectorName
+}
diff --git a/pkg/formats/format.go b/pkg/formats/format.go
index 3253e28..37c5d50 100644
--- a/pkg/formats/format.go
+++ b/pkg/formats/format.go
@@ -6,6 +6,7 @@ import (
 	"github.com/kentik/ktranslate/pkg/eggs/logger"
 
 	"github.com/kentik/ktranslate/pkg/formats/avro"
+	"github.com/kentik/ktranslate/pkg/formats/elasticsearch"
 	"github.com/kentik/ktranslate/pkg/formats/influx"
 	"github.com/kentik/ktranslate/pkg/formats/json"
 	"github.com/kentik/ktranslate/pkg/formats/netflow"
@@ -25,21 +26,24 @@ type Formatter interface {
 type Format string
 
 const (
-	FORMAT_AVRO      Format = "avro"
-	FORMAT_JSON             = "json"
-	FORMAT_JSON_FLAT        = "flat_json"
-	FORMAT_NETFLOW          = "netflow"
-	FORMAT_INFLUX           = "influx"
-	FORMAT_PROM             = "prometheus"
-	FORMAT_NR               = "new_relic"
-	FORMAT_NRM              = "new_relic_metric"
-	FORMAT_SPLUNK           = "splunk"
+	FORMAT_AVRO          Format = "avro"
+	FORMAT_ELASTICSEARCH        = "elasticsearch"
+	FORMAT_JSON                 = "json"
+	FORMAT_JSON_FLAT            = "flat_json"
+	FORMAT_NETFLOW              = "netflow"
+	FORMAT_INFLUX               = "influx"
+	FORMAT_PROM                 = "prometheus"
+	FORMAT_NR                   = "new_relic"
+	FORMAT_NRM                  = "new_relic_metric"
+	FORMAT_SPLUNK               = "splunk"
 )
 
 func NewFormat(format Format, log logger.Underlying, compression kt.Compression) (Formatter, error) {
 	switch format {
 	case FORMAT_AVRO:
 		return avro.NewFormat(log, compression)
+	case FORMAT_ELASTICSEARCH:
+		return elasticsearch.NewFormat(log, compression)
 	case FORMAT_JSON:
 		return json.NewFormat(log, compression, false)
 	case FORMAT_NETFLOW:
diff --git a/pkg/formats/nrm/nrm.go b/pkg/formats/nrm/nrm.go
index 4e60dad..d9ecbc1 100644
--- a/pkg/formats/nrm/nrm.go
+++ b/pkg/formats/nrm/nrm.go
@@ -505,12 +505,24 @@ func (f *NRMFormat) fromSnmpInterfaceMetric(in *kt.JCHF) []NRMetric {
 			if util.DropOnFilter(attrNew, f.lastMetadata[in.DeviceName], true) {
 				continue // This Metric isn't in the white list so lets drop it.
 			}
-			ms = append(ms, NRMetric{
-				Name:       "kentik.snmp." + m,
-				Type:       NR_GAUGE_TYPE,
-				Value:      int64(in.CustomBigInt[m]),
-				Attributes: attrNew,
-			})
+			switch name.Format {
+			case kt.CountMetric:
+				ms = append(ms, NRMetric{
+					Name:       "kentik.snmp." + m,
+					Type:       NR_COUNT_TYPE,
+					Value:      int64(in.CustomBigInt[m]),
+					Attributes: attrNew,
+					Interval:   in.CustomBigInt["Uptime"] * 10, // Values are in 100's of a second, so multiply by 10 to get milliseconds.
+				})
+			default:
+				ms = append(ms, NRMetric{
+					Name:       "kentik.snmp." + m,
+					Type:       NR_GAUGE_TYPE,
+					Value:      int64(in.CustomBigInt[m]),
+					Attributes: attrNew,
+					Interval:   in.CustomBigInt["Uptime"] * 10, // Values are in 100's of a second, so multiply by 10 to get milliseconds.
+				})
+			}
 		}
 	}
 
@@ -519,7 +531,7 @@ func (f *NRMFormat) fromSnmpInterfaceMetric(in *kt.JCHF) []NRMetric {
 		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 {
 						attrNew := copyAttrForSnmp(attr, "IfInUtilization", kt.MetricInfo{Oid: "computed", Mib: "computed", Profile: profileName, Table: "if"}, f.lastMetadata[in.DeviceName])
 						if !util.DropOnFilter(attrNew, f.lastMetadata[in.DeviceName], true) {
@@ -537,7 +549,7 @@ func (f *NRMFormat) fromSnmpInterfaceMetric(in *kt.JCHF) []NRMetric {
 		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 {
 						attrNew := copyAttrForSnmp(attr, "IfOutUtilization", kt.MetricInfo{Oid: "computed", Mib: "computed", Profile: profileName, Table: "if"}, f.lastMetadata[in.DeviceName])
 						if !util.DropOnFilter(attrNew, f.lastMetadata[in.DeviceName], true) {
@@ -646,6 +658,13 @@ func copyAttrForSnmp(attr map[string]interface{}, metricName string, name kt.Met
 		"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 metricName != "Uptime" { // Only allow Sys* attributes on uptime.
 			if strings.HasPrefix(k, "Sys") || k == "src_addr" {
diff --git a/pkg/inputs/snmp/metadata/device_metadata.go b/pkg/inputs/snmp/metadata/device_metadata.go
index 98c01f1..1a309b3 100644
--- a/pkg/inputs/snmp/metadata/device_metadata.go
+++ b/pkg/inputs/snmp/metadata/device_metadata.go
@@ -191,6 +191,12 @@ func getTable(log logger.ContextL, g *gosnmp.GoSNMP, oid string, mib *kt.Mib, md
 			case net.IP:
 				md.Tables[idx].Customs[oidName] = kt.NewMetaValue(mib, val.String(), 0)
 			}
+		case gosnmp.ObjectIdentifier:
+			value := string(variable.Value.(string))
+			if mib.Conversion != "" { // Adjust for any hard coded values here.
+				_, value = snmp_util.GetFromConv(variable, mib.Conversion, log)
+			}
+			md.Tables[idx].Customs[oidName] = kt.NewMetaValue(mib, value, 0)
 		default:
 			// Try to just use as a number
 			md.Tables[idx].Customs[oidName] = kt.NewMetaValue(mib, "", gosnmp.ToBigInt(variable.Value).Int64())
diff --git a/pkg/inputs/snmp/metadata/interface_metadata.go b/pkg/inputs/snmp/metadata/interface_metadata.go
index 5a699d3..a1db593 100644
--- a/pkg/inputs/snmp/metadata/interface_metadata.go
+++ b/pkg/inputs/snmp/metadata/interface_metadata.go
@@ -79,31 +79,41 @@ var (
 )
 
 type InterfaceMetadata struct {
-	log logger.ContextL
+	log           logger.ContextL
+	mibs          map[string]*kt.Mib
+	interfaceOids *orderedmap.OrderedMap
 }
 
 func NewInterfaceMetadata(interfaceMetadataMibs map[string]*kt.Mib, log logger.ContextL) *InterfaceMetadata {
+	mibs := map[string]*kt.Mib{}
+	m := orderedmap.NewOrderedMap()
+
+	// Copy the global values into this map which is per device.
+	for el := SNMP_Interface_OIDS.Front(); el != nil; el = el.Next() {
+		m.Set(el.Key, el.Value)
+	}
+
 	if len(interfaceMetadataMibs) > 0 {
 		oids := getFromCustomMap(interfaceMetadataMibs)
 		for _, oid := range oids {
-			_, ok := SNMP_Interface_OIDS.Get(oid)
+			_, ok := m.Get(oid)
 			if !ok {
 				mib := interfaceMetadataMibs[oid]
-				name := mib.Name
-				if mib.Tag != "" {
-					name = mib.Tag
-				}
+				name := mib.GetName()
 				if strings.HasPrefix(name, "if") {
 					name = name[2:]
 				}
 				log.Infof("Adding custom interface metadata oid: %s -> %s", oid, name)
-				SNMP_Interface_OIDS.Set(oid, name)
+				m.Set(oid, name)
+				mibs[name] = mib
 			}
 		}
 	}
 
 	return &InterfaceMetadata{
-		log: log,
+		log:           log,
+		mibs:          mibs,
+		interfaceOids: m,
 	}
 }
 
@@ -118,9 +128,10 @@ func (im *InterfaceMetadata) Poll(ctx context.Context, conf *kt.SnmpDeviceConfig
 	// map from ifDescr to interface definition (same structs as in intLine)
 	interfacesByDescription := make(map[string]*kt.InterfaceData)
 
-	for el := SNMP_Interface_OIDS.Front(); el != nil; el = el.Next() {
+	for el := im.interfaceOids.Front(); el != nil; el = el.Next() {
 		oidVal := el.Key.(string)
 		oidName := el.Value.(string)
+		mib := im.mibs[oidName]
 
 		results, err := snmp_util.WalkOID(ctx, conf, oidVal, server, im.log, "Interface")
 		if err != nil {
@@ -329,9 +340,19 @@ func (im *InterfaceMetadata) Poll(ctx context.Context, conf *kt.SnmpDeviceConfig
 					switch variable.Type {
 					case gosnmp.OctetString:
 						data.ExtraInfo[oidName] = string(variable.Value.([]byte))
+					case gosnmp.ObjectIdentifier:
+						data.ExtraInfo[oidName] = string(variable.Value.(string))
 					case gosnmp.Integer:
 						val := gosnmp.ToBigInt(variable.Value).Uint64()
-						data.ExtraInfo[oidName] = strconv.Itoa(int(val))
+						if mib != nil && mib.EnumRev != nil {
+							if ev, ok := mib.EnumRev[int64(val)]; ok {
+								data.ExtraInfo[oidName] = ev
+							} else {
+								data.ExtraInfo[oidName] = strconv.Itoa(int(val))
+							}
+						} else {
+							data.ExtraInfo[oidName] = strconv.Itoa(int(val))
+						}
 					}
 				}
 			}
diff --git a/pkg/inputs/snmp/metadata/poll.go b/pkg/inputs/snmp/metadata/poll.go
index 8fa9e3a..e1f6980 100644
--- a/pkg/inputs/snmp/metadata/poll.go
+++ b/pkg/inputs/snmp/metadata/poll.go
@@ -33,7 +33,7 @@ type Poller struct {
 }
 
 const (
-	DEFUALT_INTERVAL = 30 * 60 * time.Second // Run every 30 min.
+	DEFAULT_INTERVAL = 30 * 60 * time.Second // Run every 30 min.
 	vendorPrefix     = ".1.3.6.1.4.1."
 )
 
@@ -85,7 +85,7 @@ func NewPoller(server *gosnmp.GoSNMP, gconf *kt.SnmpGlobalConfig, conf *kt.SnmpD
 		conf:                  conf,
 		log:                   log,
 		server:                server,
-		interval:              DEFUALT_INTERVAL,
+		interval:              DEFAULT_INTERVAL,
 		interfaceMetadata:     NewInterfaceMetadata(interfaceMetadataMibs, log),
 		gotDeviceMetadata:     false,
 		jchfChan:              jchfChan,
diff --git a/pkg/inputs/snmp/metrics/device_metrics.go b/pkg/inputs/snmp/metrics/device_metrics.go
index f6e56a2..9a64cb3 100644
--- a/pkg/inputs/snmp/metrics/device_metrics.go
+++ b/pkg/inputs/snmp/metrics/device_metrics.go
@@ -7,7 +7,6 @@ import (
 	"strings"
 	"time"
 
-	gp "github.com/go-ping/ping"
 	"github.com/gosnmp/gosnmp"
 
 	"github.com/kentik/ktranslate/pkg/eggs/logger"
@@ -35,10 +34,7 @@ func NewDeviceMetrics(gconf *kt.SnmpGlobalConfig, conf *kt.SnmpDeviceConfig, met
 			if !strings.HasPrefix(noid, ".") {
 				noid = "." + noid
 			}
-			oidName := m.Name
-			if m.Tag != "" {
-				oidName = m.Tag
-			}
+			oidName := m.GetName()
 			log.Infof("Adding device metric %s -> %s", noid, oidName)
 			conf.DeviceOids[noid] = m
 		}
@@ -123,12 +119,6 @@ func (dm *DeviceMetrics) pollFromConfig(ctx context.Context, server *gosnmp.GoSN
 		}
 	}
 
-	// Get ping stats here if we have a service running.
-	var stats *gp.Statistics
-	if pinger != nil {
-		stats = pinger.Statistics()
-	}
-
 	// Map back into types we know about.
 	metricsFound := map[string]kt.MetricInfo{"Uptime": kt.MetricInfo{Oid: sysUpTime, Profile: dm.profileName}}
 	for _, variable := range results {
@@ -165,7 +155,7 @@ func (dm *DeviceMetrics) pollFromConfig(ctx context.Context, server *gosnmp.GoSN
 		oidName := mib.GetName()
 
 		dmr := assureDeviceMetrics(m, idx)
-		metricsFound[oidName] = kt.MetricInfo{Oid: mib.Oid, Mib: mib.Mib, Profile: dm.profileName, Table: mib.Table}
+		metricsFound[oidName] = kt.MetricInfo{Oid: mib.Oid, Mib: mib.Mib, Profile: dm.profileName, Table: mib.Table, PollDur: mib.PollDur}
 		switch variable.Type {
 		case gosnmp.OctetString, gosnmp.BitString:
 			value := string(variable.Value.([]byte))
@@ -239,18 +229,18 @@ func (dm *DeviceMetrics) pollFromConfig(ctx context.Context, server *gosnmp.GoSN
 				memoryTotal := memoryFree + memoryUsed
 				if memoryTotal > 0 {
 					dst.CustomBigInt["MemoryUtilization"] = int64(float32(memoryUsed) / float32(memoryTotal) * 100)
-					dst.CustomMetrics["MemoryUtilization"] = kt.MetricInfo{Oid: "computed", Mib: "computed", Profile: dm.profileName}
+					dst.CustomMetrics["MemoryUtilization"] = dst.CustomMetrics["MemoryFree"]
 				}
 			} else if okt && okf {
 				memoryUsed = memoryTotal - memoryFree
 				if memoryTotal > 0 {
 					dst.CustomBigInt["MemoryUtilization"] = int64(float32(memoryUsed) / float32(memoryTotal) * 100)
-					dst.CustomMetrics["MemoryUtilization"] = kt.MetricInfo{Oid: "computed", Mib: "computed", Profile: dm.profileName}
+					dst.CustomMetrics["MemoryUtilization"] = dst.CustomMetrics["MemoryFree"]
 				}
 			} else if okt && oku {
 				if memoryTotal > 0 {
 					dst.CustomBigInt["MemoryUtilization"] = int64(float32(memoryUsed) / float32(memoryTotal) * 100)
-					dst.CustomMetrics["MemoryUtilization"] = kt.MetricInfo{Oid: "computed", Mib: "computed", Profile: dm.profileName}
+					dst.CustomMetrics["MemoryUtilization"] = dst.CustomMetrics["MemoryTotal"]
 				}
 			}
 		}
@@ -273,13 +263,20 @@ func (dm *DeviceMetrics) pollFromConfig(ctx context.Context, server *gosnmp.GoSN
 		dst.CustomStr["profile_message"] = kt.DefaultProfileMessage
 	}
 
-	if stats != nil {
-		dst.CustomBigInt["MinRttMs"] = stats.MinRtt.Microseconds()
-		dst.CustomMetrics["MinRttMs"] = kt.MetricInfo{Oid: "computed", Mib: "computed", Format: kt.FloatMS, Profile: dm.profileName, Type: "ping"}
-		dst.CustomBigInt["MaxRttMs"] = stats.MaxRtt.Microseconds()
-		dst.CustomMetrics["MaxRttMs"] = kt.MetricInfo{Oid: "computed", Mib: "computed", Format: kt.FloatMS, Profile: dm.profileName, Type: "ping"}
-		dst.CustomBigInt["AvgRttMs"] = stats.AvgRtt.Microseconds()
-		dst.CustomMetrics["AvgRttMs"] = kt.MetricInfo{Oid: "computed", Mib: "computed", Format: kt.FloatMS, Profile: dm.profileName, Type: "ping"}
+	// If we are running an active check, add it in here now.
+	if pinger != nil {
+		pings, err := dm.GetPingStats(ctx, pinger)
+		if err != nil {
+			return nil, err
+		}
+		if len(pings) > 0 {
+			dst.CustomBigInt["MinRttMs"] = pings[0].CustomBigInt["MinRttMs"]
+			dst.CustomMetrics["MinRttMs"] = pings[0].CustomMetrics["MinRttMs"]
+			dst.CustomBigInt["MaxRttMs"] = pings[0].CustomBigInt["MaxRttMs"]
+			dst.CustomMetrics["MaxRttMs"] = pings[0].CustomMetrics["MaxRttMs"]
+			dst.CustomBigInt["AvgRttMs"] = pings[0].CustomBigInt["AvgRttMs"]
+			dst.CustomMetrics["AvgRttMs"] = pings[0].CustomMetrics["AvgRttMs"]
+		}
 	}
 
 	flows = append(flows, dst)
@@ -327,3 +324,29 @@ func (dm *DeviceMetrics) GetStatusFlows() []*kt.JCHF {
 	dst.CustomStr[kt.StringPrefix+"PollingHealth"] = kt.SNMP_STATUS_MAP[dst.CustomBigInt["PollingHealth"]]
 	return []*kt.JCHF{dst}
 }
+
+func (dm *DeviceMetrics) GetPingStats(ctx context.Context, pinger *ping.Pinger) ([]*kt.JCHF, error) {
+	if pinger == nil {
+		return nil, nil
+	}
+
+	stats := pinger.Statistics()
+	dst := kt.NewJCHF()
+	dst.CustomStr = map[string]string{}
+	dst.CustomInt = map[string]int32{}
+	dst.CustomBigInt = map[string]int64{}
+	dst.EventType = kt.KENTIK_EVENT_SNMP_DEV_METRIC
+	dst.Provider = dm.conf.Provider
+	dst.DeviceName = dm.conf.DeviceName
+	dst.SrcAddr = dm.conf.DeviceIP
+	dst.Timestamp = time.Now().Unix()
+	dst.CustomMetrics = map[string]kt.MetricInfo{}
+	dst.CustomBigInt["MinRttMs"] = stats.MinRtt.Microseconds()
+	dst.CustomMetrics["MinRttMs"] = kt.MetricInfo{Oid: "computed", Mib: "computed", Format: kt.FloatMS, Profile: dm.profileName, Type: "ping"}
+	dst.CustomBigInt["MaxRttMs"] = stats.MaxRtt.Microseconds()
+	dst.CustomMetrics["MaxRttMs"] = kt.MetricInfo{Oid: "computed", Mib: "computed", Format: kt.FloatMS, Profile: dm.profileName, Type: "ping"}
+	dst.CustomBigInt["AvgRttMs"] = stats.AvgRtt.Microseconds()
+	dst.CustomMetrics["AvgRttMs"] = kt.MetricInfo{Oid: "computed", Mib: "computed", Format: kt.FloatMS, Profile: dm.profileName, Type: "ping"}
+
+	return []*kt.JCHF{dst}, nil
+}
diff --git a/pkg/inputs/snmp/metrics/interface_metrics.go b/pkg/inputs/snmp/metrics/interface_metrics.go
index 889c918..55d38ef 100644
--- a/pkg/inputs/snmp/metrics/interface_metrics.go
+++ b/pkg/inputs/snmp/metrics/interface_metrics.go
@@ -32,6 +32,9 @@ const (
 	SNMP_ifHCInMulticastPkts  = "ifHCInMulticastPkts"
 	SNMP_ifHCInBroadcastPkts  = "ifHCInBroadcastPkts"
 
+	SNMP_ifInErrorsPercent  = "ifInErrorPercent"
+	SNMP_ifOutErrorsPercent = "ifOutErrorPercent"
+
 	AllDeviceInterface = "device"
 	Uptime             = "Uptime"
 )
@@ -248,11 +251,22 @@ func (im *InterfaceMetrics) convertToCHF(deltas map[string]map[string]uint64) []
 			}
 			dst.CustomBigInt[k] = int64(v)
 			if mib != nil {
-				metrics[k] = kt.MetricInfo{Oid: im.nameOidMap[k], Mib: mib.Mib, Profile: im.profileName, Table: mib.Table}
+				metrics[k] = kt.MetricInfo{Oid: im.nameOidMap[k], Mib: mib.Mib, Profile: im.profileName, Table: mib.Table, Format: kt.CountMetric, PollDur: mib.PollDur}
 			} else {
-				metrics[k] = kt.MetricInfo{Oid: im.nameOidMap[k], Profile: im.profileName}
+				metrics[k] = kt.MetricInfo{Oid: im.nameOidMap[k], Profile: im.profileName, Format: kt.CountMetric}
 			}
 		}
+
+		// Drop in Error %s here if appicable.
+		if dst.CustomBigInt[SNMP_ifHCInUcastPkts] > 0 {
+			dst.CustomBigInt[SNMP_ifInErrorsPercent] = int64(float64(dst.CustomBigInt[SNMP_ifInErrors]) / float64(dst.CustomBigInt[SNMP_ifHCInUcastPkts]) * 100.)
+			metrics[SNMP_ifInErrorsPercent] = kt.MetricInfo{Profile: im.profileName, Format: kt.GaugeMetric}
+		}
+		if dst.CustomBigInt[SNMP_ifHCOutUcastPkts] > 0 {
+			dst.CustomBigInt[SNMP_ifOutErrorsPercent] = int64(float64(dst.CustomBigInt[SNMP_ifOutErrors]) / float64(dst.CustomBigInt[SNMP_ifHCOutUcastPkts]) * 100.)
+			metrics[SNMP_ifOutErrorsPercent] = kt.MetricInfo{Profile: im.profileName, Format: kt.GaugeMetric}
+		}
+
 		if uptimeDelta > 0 {
 			dst.CustomBigInt[Uptime] = int64(uptimeDelta)
 		}
diff --git a/pkg/inputs/snmp/metrics/poll.go b/pkg/inputs/snmp/metrics/poll.go
index 88e4a22..514129a 100644
--- a/pkg/inputs/snmp/metrics/poll.go
+++ b/pkg/inputs/snmp/metrics/poll.go
@@ -52,7 +52,12 @@ func NewPoller(server *gosnmp.GoSNMP, gconf *kt.SnmpGlobalConfig, conf *kt.SnmpD
 	// If there's a profile passed in, look at the mibs set for this.
 	var deviceMetricMibs, interfaceMetricMibs map[string]*kt.Mib
 	if profile != nil {
-		deviceMetricMibs, interfaceMetricMibs = profile.GetMetrics(gconf.MibsEnabled, counterTimeSec)
+		minCounterTime := counterTimeSec
+		deviceMetricMibs, interfaceMetricMibs, minCounterTime = profile.GetMetrics(gconf.MibsEnabled, counterTimeSec)
+		if counterTimeSec != minCounterTime {
+			counterTimeSec = minCounterTime
+			log.Warnf("%d poll time adjusting to new one base on mibs", counterTimeSec)
+		}
 	}
 
 	poller := Poller{
@@ -66,7 +71,7 @@ func NewPoller(server *gosnmp.GoSNMP, gconf *kt.SnmpGlobalConfig, conf *kt.SnmpD
 		dropIfOutside:    dropIfOutside,
 	}
 
-	if gconf.RunPing || conf.RunPing {
+	if gconf.RunPing || conf.RunPing || conf.PingOnly {
 		p, err := ping.NewPinger(log, conf.DeviceIP, time.Duration(counterTimeSec)*time.Second)
 		if err != nil {
 			log.Errorf("Cannot setup ping service for %s -> %s: %v", err, conf.DeviceIP, conf.DeviceName)
@@ -178,3 +183,37 @@ func (p *Poller) Poll(ctx context.Context) ([]*kt.JCHF, error) {
 
 	return flows, nil
 }
+
+// Simpler loop which only runs on ping data, no actual snmp polling.
+func (p *Poller) StartPingOnlyLoop(ctx context.Context) {
+	// Problem is, SNMP counter polls take some time, and the time varies widely from device to device, based on number of interfaces and
+	// round-trip-time to the device.  So we're going to divide each aligned five minute chunk into two periods: an initial period over which
+	// to jitter the devices, and the rest of the five-minute chunk to actually do the counter-polling.  For any device whose counters we can walk
+	// in less than (5 minutes - jitter period), we should be able to guarantee exactly one datapoint per aligned five-minute chunk.
+	counterAlignment := time.Duration(p.counterTimeSec) * time.Second
+	jitterWindow := 15 * time.Second
+	firstCollection := time.Now().Truncate(counterAlignment).Add(counterAlignment).Add(time.Duration(rand.Int63n(int64(jitterWindow))))
+	counterCheck := tick.NewFixedTimer(firstCollection, counterAlignment)
+
+	p.log.Infof("snmpPingOnly: First run will be at %v. Running every %v, drop=%v", firstCollection, counterAlignment, p.dropIfOutside)
+
+	go func() {
+		for {
+			select {
+			case _ = <-counterCheck.C:
+				flows, err := p.deviceMetrics.GetPingStats(ctx, p.pinger)
+				if err != nil {
+					p.log.Warnf("There was an error when getting ping stats: %v.", err)
+					continue
+				}
+
+				// Send data on.
+				p.jchfChan <- flows
+
+			case <-ctx.Done():
+				p.log.Infof("Metrics PingOnly Done")
+				return
+			}
+		}
+	}()
+}
diff --git a/pkg/inputs/snmp/mibs/profile.go b/pkg/inputs/snmp/mibs/profile.go
index 70d8c1c..0123f60 100644
--- a/pkg/inputs/snmp/mibs/profile.go
+++ b/pkg/inputs/snmp/mibs/profile.go
@@ -318,9 +318,10 @@ func (p *Profile) DumpOids(log logger.ContextL) {
 // IF-MIB | ifXTable monotonic_count { } This is an interface metric because it starts with a if
 // UDP-MIB |  monotonic_count {1.3.6.1.2.1.7.8.0 udpHCInDatagrams} This is a device metrics because it does't, but still is a counter.
 
-func (p *Profile) GetMetrics(enabledMibs []string, counterTimeSec int) (map[string]*kt.Mib, map[string]*kt.Mib) {
+func (p *Profile) GetMetrics(enabledMibs []string, counterTimeSec int) (map[string]*kt.Mib, map[string]*kt.Mib, int) {
 	deviceMetrics := map[string]*kt.Mib{}
 	interfaceMetrics := map[string]*kt.Mib{}
+	minCounterTime := counterTimeSec
 
 	enabled := map[string]bool{}
 	enabledTables := map[string]map[string]bool{}
@@ -381,9 +382,11 @@ func (p *Profile) GetMetrics(enabledMibs []string, counterTimeSec int) (map[stri
 			}
 			if metric.Symbol.PollTime > 0 {
 				if counterTimeSec > metric.Symbol.PollTime {
-					p.Errorf("SNMP: mib poll time of %d is less than device time of %d. Ignoring this custom time.", metric.Symbol.PollTime, counterTimeSec)
+					p.Infof("SNMP: %s mib poll time of %d is less than device time of %d. Switching to using this interval.", metric.Symbol.Name, metric.Symbol.PollTime, counterTimeSec)
+					minCounterTime = metric.Symbol.PollTime
+					mib.PollDur = time.Duration(metric.Symbol.PollTime-kt.PollAdjustTime) * time.Second
 				} else {
-					mib.PollDur = time.Duration(metric.Symbol.PollTime) * time.Second
+					mib.PollDur = time.Duration(metric.Symbol.PollTime-kt.PollAdjustTime) * time.Second
 					p.Infof("SNMP: Custom poll time of %v for %s", mib.PollDur, mib.Name)
 				}
 			}
@@ -419,9 +422,11 @@ func (p *Profile) GetMetrics(enabledMibs []string, counterTimeSec int) (map[stri
 			}
 			if s.PollTime > 0 {
 				if counterTimeSec > s.PollTime {
-					p.Errorf("SNMP: mib poll time of %d is less than device time of %d. Ignoring this custom time.", s.PollTime, counterTimeSec)
+					p.Infof("SNMP: %s mib poll time of %d is less than device time of %d. Switching to using this interval.", s.Name, s.PollTime, counterTimeSec)
+					minCounterTime = s.PollTime
+					mib.PollDur = time.Duration(s.PollTime-kt.PollAdjustTime) * time.Second
 				} else {
-					mib.PollDur = time.Duration(s.PollTime) * time.Second
+					mib.PollDur = time.Duration(s.PollTime-kt.PollAdjustTime) * time.Second
 					p.Infof("SNMP: Custom poll time of %v for %s", mib.PollDur, mib.Name)
 				}
 			}
@@ -438,10 +443,10 @@ func (p *Profile) GetMetrics(enabledMibs []string, counterTimeSec int) (map[stri
 	}
 
 	// If we have any duplicate names, keep the longest OID.
-	prune(deviceMetrics)
-	prune(interfaceMetrics)
+	prune(deviceMetrics, counterTimeSec, minCounterTime)
+	prune(interfaceMetrics, counterTimeSec, minCounterTime)
 
-	return deviceMetrics, interfaceMetrics
+	return deviceMetrics, interfaceMetrics, minCounterTime
 }
 
 func (p *Profile) GetMetadata(enabledMibs []string) (map[string]*kt.Mib, map[string]*kt.Mib) {
@@ -547,8 +552,8 @@ func (p *Profile) GetMetadata(enabledMibs []string) (map[string]*kt.Mib, map[str
 	}
 
 	// If we have any duplicate names, keep the longest OID.
-	prune(deviceMetadata)
-	prune(interfaceMetadata)
+	prune(deviceMetadata, 0, 0)
+	prune(interfaceMetadata, 0, 0)
 
 	return deviceMetadata, interfaceMetadata
 }
@@ -682,7 +687,7 @@ func newProfileFromApc(ap *apc.APC, file string, log logger.ContextL) []*Profile
 }
 
 // Sometimes the same tag can be in multiple mibs. Run down the list and keep the longest oid if there are any collisions.
-func prune(mibs map[string]*kt.Mib) {
+func prune(mibs map[string]*kt.Mib, counterTimeSec int, minCounterTime int) {
 	seenNames := map[string]string{}
 	fromExtended := map[string]bool{}
 
@@ -713,10 +718,15 @@ func prune(mibs map[string]*kt.Mib) {
 	}
 
 	// Lastly, delete if we're not keeping.
-	for oid, _ := range mibs {
+	for oid, mib := range mibs {
 		if !keepNames[oid] {
 			delete(mibs, oid)
 		}
+
+		// Also, if we are adjusting to run faster, set the duration on any non changed mibs to the original duration.
+		if mib.PollDur == 0 && counterTimeSec != minCounterTime {
+			mib.PollDur = time.Duration(counterTimeSec-kt.PollAdjustTime) * time.Second
+		}
 	}
 }
 
diff --git a/pkg/inputs/snmp/snmp.go b/pkg/inputs/snmp/snmp.go
index 8ea2103..065c15f 100644
--- a/pkg/inputs/snmp/snmp.go
+++ b/pkg/inputs/snmp/snmp.go
@@ -150,6 +150,12 @@ func runSnmpPolling(ctx context.Context, snmpFile string, jchfChan chan []*kt.JC
 		log.Infof("Client SNMP: Running SNMP for %s on %s (type=%s)", device.DeviceName, device.DeviceIP, device.Provider)
 		metrics.Mux.Lock()
 		nm := kt.NewSnmpDeviceMetric(registry, device.DeviceName)
+
+		// Check for duplicate device names here.
+		if _, ok := metrics.Devices[device.DeviceName]; ok {
+			log.Errorf("Duplicate device name detected (%s). Is this a misconfiguration?", device.DeviceName)
+		}
+
 		metrics.Devices[device.DeviceName] = nm
 		metrics.Mux.Unlock()
 		cl := logger.NewSubContextL(logger.SContext{S: device.DeviceName}, log)
@@ -199,6 +205,10 @@ func launchSnmpTrap(conf *kt.SnmpConfig, jchfChan chan []*kt.JCHF, metrics *kt.S
 }
 
 func launchSnmp(ctx context.Context, conf *kt.SnmpGlobalConfig, device *kt.SnmpDeviceConfig, jchfChan chan []*kt.JCHF, connectTimeout time.Duration, retries int, metrics *kt.SnmpDeviceMetric, profile *mibs.Profile, log logger.ContextL) error {
+	// Sometimes this device is pinging only. In this case, start the ping loop and return.
+	if device.PingOnly {
+		return launchPingOnly(ctx, conf, device, jchfChan, connectTimeout, retries, metrics, profile, log)
+	}
 
 	// We need two of these, to avoid concurrent access by the two pollers.
 	// gosnmp isn't real clear on its approach to concurrency, but it seems
@@ -334,6 +344,28 @@ func parseConfig(file string, log logger.ContextL) (*kt.SnmpConfig, error) {
 	return &ms, nil
 }
 
+/**
+Handle the case where we're only doing a ping loop of a device.
+*/
+func launchPingOnly(ctx context.Context, conf *kt.SnmpGlobalConfig, device *kt.SnmpDeviceConfig, jchfChan chan []*kt.JCHF, connectTimeout time.Duration, retries int, metrics *kt.SnmpDeviceMetric, profile *mibs.Profile, log logger.ContextL) error {
+	metricsServer, err := snmp_util.InitSNMP(device, connectTimeout, retries, "", log)
+	if err != nil {
+		log.Warnf("There was an error when starting SNMP interface component -- %v.", err)
+		metrics.Fail.Update(kt.SNMP_BAD)
+		return err
+	}
+
+	metricPoller := snmp_metrics.NewPoller(metricsServer, conf, device, jchfChan, metrics, profile, log)
+
+	// We've now done everything we can do synchronously -- return to the client initialization
+	// code, and do everything else in the background
+	go func() {
+		metricPoller.StartPingOnlyLoop(ctx)
+	}()
+
+	return nil
+}
+
 // Public wrapper for calling this other places.
 func ParseConfig(file string, log logger.ContextL) (*kt.SnmpConfig, error) {
 	return parseConfig(file, log)
diff --git a/pkg/inputs/snmp/util/util.go b/pkg/inputs/snmp/util/util.go
index 1eee242..2187029 100644
--- a/pkg/inputs/snmp/util/util.go
+++ b/pkg/inputs/snmp/util/util.go
@@ -8,6 +8,7 @@ import (
 	"encoding/hex"
 	"fmt"
 	"net"
+	"regexp"
 	"strconv"
 	"strings"
 	"time"
@@ -24,6 +25,7 @@ const (
 	CONV_HEXTOINT  = "hextoint"
 	CONV_HEXTOIP   = "hextoip"
 	CONV_ENGINE_ID = "engine_id"
+	CONV_REGEXP    = "regexp"
 
 	NRUserTagPrefix = "tags."
 )
@@ -33,6 +35,9 @@ var (
 
 	// Device Manufacturer aka sysDescr
 	SNMP_DEVICE_MANUFACTURER_OID = "1.3.6.1.2.1.1.1"
+
+	// Keep a cache of seen regexps seen to speed things up.
+	reCache = map[string]*regexp.Regexp{}
 )
 
 func ContainsAny(s string, substrs ...string) bool {
@@ -282,6 +287,8 @@ func GetFromConv(pdu gosnmp.SnmpPDU, conv string, log logger.ContextL) (int64, s
 				log.Errorf("invalid Endian value (%s) for hex to int conversion", endian)
 				return 0, ""
 			}
+		} else if split[0] == CONV_REGEXP && len(split) >= 2 {
+			return fromRegexp(bv, strings.Join(split[1:], ":")) // Put back together just in case RE has a : in it.
 		}
 	}
 
@@ -325,3 +332,33 @@ func engineID(bv []byte) (int64, string) {
 	}
 	return 0, string(buf[:len(buf)-1])
 }
+
+/**
+Ubiquity and maybe others can be annoying in returning a string version of CPU.
+This lets people parse it out.
+
+	agentSwitchCpuProcessTotalUtilization
+	1.3.6.1.4.1.4413.1.1.1.1.4.9.0 = STRING: "    5 Secs ( 96.3762%)   60 Secs ( 62.8549%)  300 Secs ( 25.2877%)"
+*/
+func fromRegexp(bv []byte, reg string) (int64, string) {
+	r := reCache[reg]
+	if r == nil {
+		rn, err := regexp.Compile(reg)
+		if err != nil {
+			return 0, ""
+		}
+		reCache[reg] = rn
+		r = rn
+	}
+
+	// Now, lets run some regexps.
+	res := r.FindSubmatch(bv)
+	if len(res) < 2 {
+		return 0, ""
+	}
+	ival, err := strconv.Atoi(string(res[1]))
+	if err != nil { // If we can't parse as int, return as string.
+		return 0, string(res[1])
+	}
+	return int64(ival), string(res[1]) // Parsed as int but return both just in case.
+}
diff --git a/pkg/inputs/vpc/aws/aws.go b/pkg/inputs/vpc/aws/aws.go
index 9208324..c00710d 100644
--- a/pkg/inputs/vpc/aws/aws.go
+++ b/pkg/inputs/vpc/aws/aws.go
@@ -76,6 +76,8 @@ const (
 	AWS_PKT_DST_AWS_SERVICE = "pkt-dst-aws-service"
 	AWS_FLOW_DIRECTION      = "flow-direction"
 	AWS_TRAFFIC_PATH        = "traffic-path"
+
+	AWS_VPC_TYPE = "AWS_VPC"
 )
 
 var (
@@ -468,6 +470,7 @@ func (m *AWSLogLine) ToFlow(log logger.ContextL, topo *AWSTopology) (in *kt.JCHF
 	in.CustomStr["action"] = m.Action
 	in.CustomStr["status"] = m.Status
 	in.CustomStr["kt.from"] = kt.FromLambda
+	in.CustomStr["type"] = AWS_VPC_TYPE
 	in.DeviceName = m.VPCID
 
 	if m.Sample > 0 { // Set sample rate here if we are switching.
diff --git a/pkg/kt/snmp.go b/pkg/kt/snmp.go
index e6de577..0197a98 100644
--- a/pkg/kt/snmp.go
+++ b/pkg/kt/snmp.go
@@ -77,6 +77,7 @@ type DeviceMetricsMetadata struct {
 const (
 	LOGICAL_INTERFACE  = "logical"
 	PHYSICAL_INTERFACE = "physical"
+	PollAdjustTime     = 5
 )
 
 var (
@@ -140,6 +141,7 @@ type SnmpDeviceConfig struct {
 	MibProfile             string            `yaml:"mib_profile"`
 	Provider               Provider          `yaml:"provider"`
 	FlowOnly               bool              `yaml:"flow_only,omitempty"`
+	PingOnly               bool              `yaml:"ping_only,omitempty"`
 	UserTags               map[string]string `yaml:"user_tags"`
 	DiscoveredMibs         []string          `yaml:"discovered_mibs,omitempty"`
 	PollTimeSec            int               `yaml:"poll_time_sec,omitempty"`
diff --git a/pkg/kt/types.go b/pkg/kt/types.go
index 095b983..7ef5bfd 100644
--- a/pkg/kt/types.go
+++ b/pkg/kt/types.go
@@ -34,7 +34,7 @@ const (
 	StringPrefix     = "ks_"
 	PrivateIP        = "Private IP"
 	DropMetric       = "DropMetric"
-	AdminStatus      = "ifAdminStatus"
+	AdminStatus      = "if_AdminStatus"
 )
 
 type OutputType string
@@ -84,6 +84,8 @@ const (
 	InvalidEnum    = "invalid enum"
 	DeviceTagTable = "KT_Device_Tag"
 	FromLambda     = "awslambda"
+	GaugeMetric    = "gauge"
+	CountMetric    = "count"
 )
 
 type IntId uint64
@@ -180,13 +182,14 @@ type JCHF struct {
 }
 
 type MetricInfo struct {
-	Oid     string `json:"-"`
-	Mib     string `json:"-"`
-	Name    string `json:"-"`
-	Profile string `json:"-"`
-	Table   string `json:"-"`
-	Format  string `json:"-"`
-	Type    string `json:"-"`
+	Oid     string        `json:"-"`
+	Mib     string        `json:"-"`
+	Name    string        `json:"-"`
+	Profile string        `json:"-"`
+	Table   string        `json:"-"`
+	Format  string        `json:"-"`
+	Type    string        `json:"-"`
+	PollDur time.Duration `json:"-"`
 }
 
 func (m *MetricInfo) GetType() string {
diff --git a/pkg/sinks/gcppubsub/gcppubsub.go b/pkg/sinks/gcppubsub/gcppubsub.go
new file mode 100644
index 0000000..5f84417
--- /dev/null
+++ b/pkg/sinks/gcppubsub/gcppubsub.go
@@ -0,0 +1,96 @@
+package gcppubsub
+
+import (
+	"context"
+	"flag"
+	"fmt"
+
+	"cloud.google.com/go/pubsub"
+	go_metrics "github.com/kentik/go-metrics"
+	"github.com/kentik/ktranslate/pkg/eggs/logger"
+	"github.com/kentik/ktranslate/pkg/formats"
+	"github.com/kentik/ktranslate/pkg/kt"
+)
+
+type GCPPubSub struct {
+	logger.ContextL
+	client    *pubsub.Client
+	registry  go_metrics.Registry
+	metrics   *GCPPubSubMetric
+	projectID string
+	topicID   string
+	topic     *pubsub.Topic
+}
+
+type GCPPubSubMetric struct {
+	DeliveryErr go_metrics.Meter
+	DeliveryWin go_metrics.Meter
+}
+
+var (
+	GCPPubSubProjectID = flag.String("gcp_pubsub_project_id", "", "GCP PubSub Project ID to use")
+	GCPPubSubTopic     = flag.String("gcp_pubsub_topic", "", "GCP PubSub Topic to publish to")
+)
+
+func NewSink(log logger.Underlying, registry go_metrics.Registry) (*GCPPubSub, error) {
+	return &GCPPubSub{
+		registry:  registry,
+		projectID: *GCPPubSubProjectID,
+		topicID:   *GCPPubSubTopic,
+		ContextL:  logger.NewContextLFromUnderlying(logger.SContext{S: "gcppubsub"}, log),
+		metrics: &GCPPubSubMetric{
+			DeliveryErr: go_metrics.GetOrRegisterMeter("delivery_errors_gcppubsub", registry),
+			DeliveryWin: go_metrics.GetOrRegisterMeter("delivery_wins_gcppubsub", registry),
+		},
+	}, nil
+}
+
+func (s *GCPPubSub) Init(ctx context.Context, format formats.Format, compression kt.Compression, fmtr formats.Formatter) error {
+	if s.projectID == "" {
+		return fmt.Errorf("No project ID was set for GCP PubSub. Use the -gcp_pubsub_project_id option to set one up.")
+	}
+	if s.topicID == "" {
+		return fmt.Errorf("No topic was set for GCP PubSub. Use the -gcp_pubsub_topic option to set one up.")
+	}
+	client, err := pubsub.NewClient(ctx, s.projectID)
+	if err != nil {
+		return fmt.Errorf("There was an error when creating the gcp pubsub client: %v.", err)
+	}
+	s.client = client
+	s.topic = s.client.Topic(s.topicID)
+
+	s.Infof("connected to GCP PubSub for project %s on topic %s", s.projectID, s.topicID)
+
+	return nil
+}
+
+func (s *GCPPubSub) Send(ctx context.Context, payload *kt.Output) {
+	go s.send(ctx, payload.Body)
+}
+
+func (s *GCPPubSub) send(ctx context.Context, payload []byte) {
+	res := s.topic.Publish(ctx, &pubsub.Message{Data: payload})
+	if _, err := res.Get(ctx); err != nil {
+		s.Errorf("Error publishing to GCP: %v", err)
+		s.metrics.DeliveryErr.Mark(1)
+		return
+	}
+	s.metrics.DeliveryWin.Mark(1)
+}
+
+func (s *GCPPubSub) Close() {
+	if s.topic != nil {
+		s.topic.Stop()
+	}
+
+	if s.client != nil {
+		s.client.Close()
+	}
+}
+
+func (s *GCPPubSub) HttpInfo() map[string]float64 {
+	return map[string]float64{
+		"DeliveryErr": s.metrics.DeliveryErr.Rate1(),
+		"DeliveryWin": s.metrics.DeliveryWin.Rate1(),
+	}
+}
diff --git a/pkg/sinks/sinks.go b/pkg/sinks/sinks.go
index 3a00952..6bc6bc1 100644
--- a/pkg/sinks/sinks.go
+++ b/pkg/sinks/sinks.go
@@ -11,6 +11,7 @@ import (
 	"github.com/kentik/ktranslate/pkg/kt"
 	"github.com/kentik/ktranslate/pkg/sinks/file"
 	"github.com/kentik/ktranslate/pkg/sinks/gcloud"
+	"github.com/kentik/ktranslate/pkg/sinks/gcppubsub"
 	"github.com/kentik/ktranslate/pkg/sinks/http"
 	"github.com/kentik/ktranslate/pkg/sinks/kafka"
 	"github.com/kentik/ktranslate/pkg/sinks/kentik"
@@ -42,6 +43,7 @@ const (
 	PromSink          = "prometheus"
 	S3Sink            = "s3"
 	GCloudSink        = "gcloud"
+	GCPPubSub         = "gcppubsub"
 )
 
 func NewSink(sink Sink, log logger.Underlying, registry go_metrics.Registry, tooBig chan int, conf *kt.KentikConfig, logTee chan string) (SinkImpl, error) {
@@ -66,6 +68,8 @@ func NewSink(sink Sink, log logger.Underlying, registry go_metrics.Registry, too
 		return s3.NewSink(log, registry)
 	case GCloudSink:
 		return gcloud.NewSink(log, registry)
+	case GCPPubSub:
+		return gcppubsub.NewSink(log, registry)
 	}
 	return nil, fmt.Errorf("Unknown sink %v", sink)
 }


Test patch:
diff --git a/pkg/formats/elasticsearch/elasticsearch_test.go b/pkg/formats/elasticsearch/elasticsearch_test.go
new file mode 100644
index 0000000..9e3c27a
--- /dev/null
+++ b/pkg/formats/elasticsearch/elasticsearch_test.go
@@ -0,0 +1,45 @@
+package elasticsearch
+
+import (
+	"testing"
+
+	"github.com/kentik/ktranslate/pkg/kt"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSerializeElasticsearch(t *testing.T) {
+	serBuf := make([]byte, 0)
+	assert := assert.New(t)
+
+	f, err := NewFormat(nil, kt.CompressionNone)
+	assert.NoError(err)
+
+	res, err := f.To(kt.InputTesting, serBuf)
+	assert.NoError(err)
+	assert.NotNil(res)
+
+	out, err := f.From(res)
+	assert.NoError(err)
+	assert.Equal(len(kt.InputTesting), len(out))
+	for i, _ := range out {
+		assert.Equal(kt.InputTesting[i].SrcAddr, out[i]["src_addr"])
+		assert.Equal(int(0), int(out[i]["timestamp"].(int64)))
+	}
+}
+
+func TestSerializeElasticsearchGzip(t *testing.T) {
+	serBuf := make([]byte, 0)
+	assert := assert.New(t)
+	f, err := NewFormat(nil, kt.CompressionGzip)
+	assert.NoError(err)
+	res, err := f.To(kt.InputTesting, serBuf)
+	assert.NoError(err)
+	assert.NotNil(res)
+	out, err := f.From(res)
+	assert.NoError(err)
+	assert.Equal(len(kt.InputTesting), len(out))
+	for i, _ := range out {
+		assert.Equal(kt.InputTesting[i].SrcAddr, out[i]["src_addr"])
+		assert.Equal(int(0), int(out[i]["timestamp"].(int64)))
+	}
+}
diff --git a/pkg/inputs/snmp/mibs/profile_test.go b/pkg/inputs/snmp/mibs/profile_test.go
index afe6fbc..7dc557f 100644
--- a/pkg/inputs/snmp/mibs/profile_test.go
+++ b/pkg/inputs/snmp/mibs/profile_test.go
@@ -2,6 +2,7 @@ package mibs
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 
@@ -119,7 +120,7 @@ func TestPrune(t *testing.T) {
 	mibs := map[string]*kt.Mib{
 		".1.3.6.1.4.1.9.9.48.1.1.1.5": &kt.Mib{Tag: "MemoryUsed"},
 	}
-	prune(mibs)
+	prune(mibs, 0, 0)
 	assert.Equal(t, 1, len(mibs))
 
 	mibs = map[string]*kt.Mib{
@@ -128,7 +129,7 @@ func TestPrune(t *testing.T) {
 		".1.3.6.1.4.1.9.9.109.1.1.1.1.7":  &kt.Mib{Tag: "CPU"},
 		".1.3.6.1.4.1.9.9.109.1.1.1.1.13": &kt.Mib{Name: "cpmCPUMemoryFree"},
 	}
-	prune(mibs)
+	prune(mibs, 0, 0)
 	assert.Equal(t, 3, len(mibs))
 	assert.Nil(t, mibs[".1.3.6.1.4.1.9.9.305.1.1.1"], len(mibs))
 	assert.NotNil(t, mibs[".1.3.6.1.4.1.9.9.109.1.1.1.1.7"], len(mibs))
@@ -139,10 +140,16 @@ func TestPrune(t *testing.T) {
 		".1.3.6.1.4.1.9.9.109.1.1.1.1.7":  &kt.Mib{Tag: "CPU", FromExtended: true},
 		".1.3.6.1.4.1.9.9.109.1.1.1.1.13": &kt.Mib{Name: "cpmCPUMemoryFree"},
 	}
-	prune(mibs)
+	prune(mibs, 0, 0)
 	assert.Equal(t, 3, len(mibs))
 	assert.NotNil(t, mibs[".1.3.6.1.4.1.9.9.305.1.1.1"], len(mibs))
 	assert.Nil(t, mibs[".1.3.6.1.4.1.9.9.109.1.1.1.1.7"], len(mibs))
+	assert.Equal(t, mibs[".1.3.6.1.4.1.9.9.305.1.1.1"].PollDur, time.Duration(0)*time.Second)
+
+	prune(mibs, 60, 30)
+	assert.Equal(t, 3, len(mibs))
+	assert.NotNil(t, mibs[".1.3.6.1.4.1.9.9.305.1.1.1"], len(mibs))
+	assert.Equal(t, mibs[".1.3.6.1.4.1.9.9.305.1.1.1"].PollDur, time.Duration(60-kt.PollAdjustTime)*time.Second)
 }
 
 func TestProfileName(t *testing.T) {
diff --git a/pkg/inputs/snmp/util/util_test.go b/pkg/inputs/snmp/util/util_test.go
index 2c08e89..f104c28 100644
--- a/pkg/inputs/snmp/util/util_test.go
+++ b/pkg/inputs/snmp/util/util_test.go
@@ -175,3 +175,19 @@ func TestEngineID(t *testing.T) {
 		assert.Equal(expt, str, "%s", in)
 	}
 }
+
+func TestRegex(t *testing.T) {
+	assert := assert.New(t)
+	l := lt.NewTestContextL(logger.NilContext, t)
+
+	tests := map[int64][]byte{
+		62: []byte("    5 Secs ( 96.3762%)   60 Secs ( 62.8549%)  300 Secs ( 25.2877%)"),
+		64: []byte("    5 Secs ( 96.3762%)   60 Secs ( 64.8549%)  300 Secs ( 25.2877%)"),
+	}
+
+	for expt, in := range tests {
+		pdu := gosnmp.SnmpPDU{Value: in}
+		ival, _ := GetFromConv(pdu, CONV_REGEXP+`:60 Secs.*?(\d+)`, l)
+		assert.Equal(expt, ival, "%s", in)
+	}
+}