Repository: kentik/ktranslate
Pull request #495 (2022-12-03 00:32:37)

Wip for what a managed config system might look like
Here's a rough outline for what a managed config system might look like. 

https://github.com/kentik/ktranslate/tree/wip-managed-config/pkg/config/local 
Is an example of something which looks for a config change on a local file and restarts things if it sees this.

https://github.com/kentik/ktranslate/tree/wip-managed-config/pkg/config/nr
is a stub where you can pass data to NR. 

Thoughts here, especially on this interface?

```
type ConfigManager interface {
	Run(context.Context, func(*ktranslate.Config) error) // Run takes a context and a callback function to call whenever there is a new update to process
	DeviceDiscovery(kt.DeviceMap)                        // called whenever there is a new snmp device discovery to parse.
	Close()                                              // Called on shutdown of ktrans.
}
```



Patch:
diff --git a/cmd/ktranslate/main.go b/cmd/ktranslate/main.go
index fae331e..dc7d552 100644
--- a/cmd/ktranslate/main.go
+++ b/cmd/ktranslate/main.go
@@ -70,11 +70,11 @@ func init() {
 	flag.IntVar(&threads, "threads", 1, "Number of threads to run for processing")
 	flag.IntVar(&threadsInput, "input_threads", 1, "Number of threads to run for input processing")
 	flag.IntVar(&maxThreads, "max_threads", 1, "Dynamically grow threads up to this number")
-	flag.StringVar(&format, "format", "flat_json", "Format to convert kflow to: (json|flat_json|avro|netflow|influx|carbon|prometheus|new_relic|new_relic_metric|splunk|elasticsearch|kflow)")
+	flag.StringVar(&format, "format", "flat_json", "Format to convert kflow to: (json|flat_json|avro|netflow|influx|carbon|prometheus|new_relic|new_relic_metric|splunk|elasticsearch|kflow|ddog)")
 	flag.StringVar(&formatRollup, "format_rollup", "", "Format to convert rollups to: (json|avro|netflow|influx|prometheus|new_relic|new_relic_metric|splunk|elasticsearch|kflow)")
 	flag.StringVar(&formatMetric, "format_metric", "", "Format to convert metrics to: (json|avro|netflow|influx|prometheus|new_relic|new_relic_metric|splunk|elasticsearch|kflow)")
 	flag.StringVar(&compression, "compression", "none", "compression algo to use (none|gzip|snappy|deflate|null)")
-	flag.StringVar(&sinks, "sinks", "stdout", "List of sinks to send data to. Options: (kafka|stdout|new_relic|kentik|net|http|splunk|prometheus|file|s3|gcloud)")
+	flag.StringVar(&sinks, "sinks", "stdout", "List of sinks to send data to. Options: (kafka|stdout|new_relic|kentik|net|http|splunk|prometheus|file|s3|gcloud|ddog)")
 	flag.IntVar(&maxFlows, "max_flows_per_message", 10000, "Max number of flows to put in each emitted message")
 	flag.IntVar(&dumpRollups, "rollup_interval", 0, "Export timer for rollups in seconds")
 	flag.BoolVar(&rollupAndAlpha, "rollup_and_alpha", false, "Send both rollups and alpha inputs to sinks")
@@ -401,6 +401,8 @@ func applyFlags(cfg *ktranslate.Config) error {
 				cfg.EnableHTTPInput = v
 			case "enricher":
 				cfg.EnricherURL = val
+			case "ddog_url":
+				cfg.DDogSink.URL = val
 			// pkg/maps/file
 			case "tag_map":
 				cfg.TagMapFile = val
@@ -425,6 +427,8 @@ func applyFlags(cfg *ktranslate.Config) error {
 			// pkg/formats/influxdb
 			case "influxdb_measurement_prefix":
 				cfg.InfluxDBFormat.MeasurementPrefix = val
+			case "influxdb_namespace_token":
+				cfg.InfluxDBFormat.NamespaceToken = val
 			// pkg/sinks/prom
 			case "prom_listen":
 				cfg.PrometheusSink.ListenAddr = val
@@ -707,6 +711,8 @@ func applyFlags(cfg *ktranslate.Config) error {
 				cfg.FlowInput.PrometheusListenAddr = val
 			case "nf.mapping":
 				cfg.FlowInput.MappingFile = val
+			case "config_provider":
+				cfg.CfgManager.ConfigImpl = val
 			// configs
 			case "config", "generate-config":
 				// ignore
diff --git a/config.go b/config.go
index 9a2f6a9..ee6399f 100644
--- a/config.go
+++ b/config.go
@@ -23,6 +23,7 @@ type NetflowFormatConfig struct {
 // InfluxDBFormatConfig is the config format for influxdb
 type InfluxDBFormatConfig struct {
 	MeasurementPrefix string
+	NamespaceToken    string
 }
 
 // PrometheusFormatConfig is the config for the prometheus format
@@ -108,6 +109,11 @@ type KentikSinkConfig struct {
 	RelayURL string
 }
 
+// DDogSinkConfig is config for the ddog sink
+type DDogSinkConfig struct {
+	URL string
+}
+
 // RollupConfig is the config for rollups
 type RollupConfig struct {
 	JoinKey string
@@ -129,7 +135,6 @@ type ServerConfig struct {
 	MetaListenAddr  string
 	OllyDataset     string
 	OllyWriteKey    string
-	MonitorConf     bool
 	CfgPath         string `yaml:"-"` // We don't want to read this directly because it comes from a flag but saved here for internal use.
 }
 
@@ -203,6 +208,12 @@ type KentikCred struct {
 	APIToken string
 }
 
+// ConfigManager is the config for how to manage configs.
+type ConfigManager struct {
+	ConfigImpl  string
+	PollTimeSec int
+}
+
 // Config is the ktranslate configuration
 type Config struct {
 	// ktranslate
@@ -270,6 +281,8 @@ type Config struct {
 	KafkaSink *KafkaSinkConfig
 	// pkg/sinks/kentik
 	KentikSink *KentikSinkConfig
+	// pkg/sinks/ddog
+	DDogSink *DDogSinkConfig
 
 	// pkg/rollup
 	Rollup *RollupConfig
@@ -292,6 +305,8 @@ type Config struct {
 	AWSVPCInput *AWSVPCInputConfig
 	// pkg/inputs/flow
 	FlowInput *FlowInputConfig
+	// pkg/config
+	CfgManager *ConfigManager
 }
 
 // DefaultConfig returns a ktranslate configuration with defaults applied
@@ -336,6 +351,7 @@ func DefaultConfig() *Config {
 		},
 		InfluxDBFormat: &InfluxDBFormatConfig{
 			MeasurementPrefix: "",
+			NamespaceToken:    ":",
 		},
 		PrometheusSink: &PrometheusSinkConfig{
 			ListenAddr: ":8082",
@@ -364,6 +380,9 @@ func DefaultConfig() *Config {
 		NewRelicMultiSink: &NewRelicMultiSinkConfig{
 			CredMap: nil,
 		},
+		DDogSink: &DDogSinkConfig{
+			URL: "",
+		},
 		FileSink: &FileSinkConfig{
 			Path:                 "./",
 			EnableImmediateWrite: false,
@@ -403,7 +422,6 @@ func DefaultConfig() *Config {
 			OllyDataset:     "",
 			OllyWriteKey:    "",
 			CfgPath:         "",
-			MonitorConf:     false,
 		},
 		API: &APIConfig{
 			DeviceFile: "",
@@ -456,6 +474,10 @@ func DefaultConfig() *Config {
 			PrometheusListenAddr: "",
 			MappingFile:          "",
 		},
+		CfgManager: &ConfigManager{
+			ConfigImpl:  "",
+			PollTimeSec: 1200,
+		},
 	}
 }
 
diff --git a/go.mod b/go.mod
index 1098b82..710ea52 100644
--- a/go.mod
+++ b/go.mod
@@ -8,7 +8,7 @@ require (
 	cloud.google.com/go/storage v1.18.2
 	github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.1.0
 	github.com/Azure/azure-sdk-for-go/sdk/keyvault/azsecrets v0.7.1
-	github.com/BurntSushi/toml v0.3.1
+	github.com/DataDog/datadog-api-client-go/v2 v2.5.0
 	github.com/aristanetworks/goeapi v0.6.0
 	github.com/aws/aws-lambda-go v1.27.0
 	github.com/aws/aws-sdk-go v1.41.19
@@ -20,6 +20,7 @@ require (
 	github.com/gorilla/mux v1.8.0
 	github.com/gosnmp/gosnmp v1.35.0
 	github.com/honeycombio/libhoney-go v1.15.6
+	github.com/imdario/mergo v0.3.13
 	github.com/influxdata/line-protocol/v2 v2.2.1
 	github.com/json-iterator/go v1.1.12
 	github.com/judwhite/go-svc v1.2.1
@@ -30,16 +31,20 @@ require (
 	github.com/liamg/furious v0.0.0-20191231090757-c295c872d6c1
 	github.com/linkedin/goavro/v2 v2.10.1
 	github.com/montanaflynn/stats v0.6.6
+	github.com/netsampler/goflow2 v1.1.1-0.20220818030102-fc42020d1b73
 	github.com/oschwald/geoip2-golang v1.5.0
+	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.11.0
 	github.com/sasha-s/go-hll v0.0.0-20180522065212-c6eb27aee351
 	github.com/segmentio/kafka-go v0.4.23
 	github.com/stretchr/testify v1.7.1
 	github.com/syndtr/goleveldb v1.0.0
+	go.starlark.net v0.0.0-20220926145019-14b050677505
 	google.golang.org/genproto v0.0.0-20220405205423-9d709892a2bf
 	google.golang.org/grpc v1.45.0
 	gopkg.in/mcuadros/go-syslog.v2 v2.3.0
 	gopkg.in/yaml.v2 v2.4.0
+	gopkg.in/yaml.v3 v3.0.0
 	zombiezen.com/go/capnproto2 v2.18.2+incompatible
 )
 
@@ -51,6 +56,7 @@ require (
 	github.com/Azure/azure-sdk-for-go/sdk/internal v1.0.0 // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/keyvault/internal v0.5.0 // indirect
 	github.com/AzureAD/microsoft-authentication-library-for-go v0.5.1 // indirect
+	github.com/DataDog/zstd v1.5.0 // indirect
 	github.com/PuerkitoBio/purell v1.1.1 // indirect
 	github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 // indirect
 	github.com/asaskevich/govalidator v0.0.0-20210307081110-f21760c49a8d // indirect
@@ -80,7 +86,6 @@ require (
 	github.com/google/gopacket v1.1.17 // indirect
 	github.com/googleapis/gax-go/v2 v2.2.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.6.0 // indirect
-	github.com/imdario/mergo v0.3.13 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/klauspost/compress v1.13.6 // indirect
@@ -93,7 +98,6 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/mostlygeek/arp v0.0.0-20170424181311-541a2129847a // indirect
-	github.com/netsampler/goflow2 v1.1.1-0.20220818030102-fc42020d1b73 // indirect
 	github.com/oklog/ulid v1.3.1 // indirect
 	github.com/onsi/ginkgo v1.16.2 // indirect
 	github.com/onsi/gomega v1.13.0 // indirect
@@ -102,7 +106,6 @@ require (
 	github.com/phayes/freeport v0.0.0-20180830031419-95f893ade6f2 // indirect
 	github.com/pierrec/lz4 v2.6.1+incompatible // indirect
 	github.com/pkg/browser v0.0.0-20210115035449-ce105d075bb4 // indirect
-	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common v0.30.0 // indirect
@@ -115,7 +118,6 @@ require (
 	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
 	go.mongodb.org/mongo-driver v1.8.4 // indirect
 	go.opencensus.io v0.23.0 // indirect
-	go.starlark.net v0.0.0-20220926145019-14b050677505 // indirect
 	golang.org/x/crypto v0.0.0-20220511200225-c6db032c6c88 // indirect
 	golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4 // indirect
 	golang.org/x/oauth2 v0.0.0-20220309155454-6242fa91716a // indirect
@@ -127,5 +129,4 @@ require (
 	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/protobuf v1.28.0 // indirect
 	gopkg.in/alexcesaro/statsd.v2 v2.0.0 // indirect
-	gopkg.in/yaml.v3 v3.0.0 // indirect
 )
diff --git a/go.sum b/go.sum
index a63f194..57e02b0 100644
--- a/go.sum
+++ b/go.sum
@@ -72,11 +72,13 @@ github.com/Azure/azure-sdk-for-go/sdk/keyvault/internal v0.5.0 h1:9cn6ICCGiWFNA/
 github.com/Azure/azure-sdk-for-go/sdk/keyvault/internal v0.5.0/go.mod h1:9V2j0jn9jDEkCkv8w/bKTNppX/d0FVA1ud77xCIP4KA=
 github.com/AzureAD/microsoft-authentication-library-for-go v0.5.1 h1:BWe8a+f/t+7KY7zH2mqygeUD0t8hNFXe08p1Pb3/jKE=
 github.com/AzureAD/microsoft-authentication-library-for-go v0.5.1/go.mod h1:Vt9sXTKwMyGcOxSmLDMnGPgqsUg7m8pe215qMLrDXw4=
-github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
-github.com/DataDog/zstd v1.4.8 h1:Rpmta4xZ/MgZnriKNd24iZMhGpP5dvUcs/uqfBapKZY=
+github.com/DataDog/datadog-api-client-go/v2 v2.5.0 h1:+FRtNnR+IZ4D5lGqEk+f0+dSP7XhKtrlHvdcnTdyny8=
+github.com/DataDog/datadog-api-client-go/v2 v2.5.0/go.mod h1:98b/MtTwSAr/yhTfhCR1oxAqQ/4tMkdrgKH7fYiDA0g=
 github.com/DataDog/zstd v1.4.8/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw=
+github.com/DataDog/zstd v1.5.0 h1:+K/VEwIAaPcHiMtQvpLD4lqW7f0Gk3xdYZmI1hD+CXo=
+github.com/DataDog/zstd v1.5.0/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw=
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
 github.com/PuerkitoBio/purell v1.1.1 h1:WEQqlqaGbrPkxLJWfBwQmfEAE1Z7ONdDLqrN38tNFfI=
 github.com/PuerkitoBio/purell v1.1.1/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0=
@@ -794,7 +796,6 @@ golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220209214540-3681064d5158/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220227234510-4e6760a101f9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220328115105-d36c6a25d886 h1:eJv7u3ksNXoLbGSKuv2s/SIO4tJVxc/A+MTpzxDgz/Q=
 golang.org/x/sys v0.0.0-20220328115105-d36c6a25d886/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 h1:0A+M6Uqn+Eje4kHMK80dtF3JCXC4ykBgQG4Fe06QRhQ=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -1070,7 +1071,6 @@ gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20200605160147-a5ece683394c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0 h1:hjy8E9ON/egN1tAYqKb61G10WtihqetD4sz2H+8nIeA=
 gopkg.in/yaml.v3 v3.0.0/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/ktranslate-package.yml b/ktranslate-package.yml
index 3f0777e..3a87e1f 100644
--- a/ktranslate-package.yml
+++ b/ktranslate-package.yml
@@ -42,6 +42,7 @@ files:
     file: config/snmp-base.yaml
     mode: "0644"
     user: ktranslate
+    keep: true
 
   "/etc/default/ktranslate.env.sample":
     file: config/ktranslate.env.sample
diff --git a/pkg/cat/kkc.go b/pkg/cat/kkc.go
index 7389c81..d0e8cc3 100644
--- a/pkg/cat/kkc.go
+++ b/pkg/cat/kkc.go
@@ -11,6 +11,7 @@ import (
 	"github.com/kentik/ktranslate"
 	"github.com/kentik/ktranslate/pkg/api"
 	"github.com/kentik/ktranslate/pkg/cat/auth"
+	cfgMngr "github.com/kentik/ktranslate/pkg/config"
 	"github.com/kentik/ktranslate/pkg/eggs/baseserver"
 	"github.com/kentik/ktranslate/pkg/eggs/kmux"
 	"github.com/kentik/ktranslate/pkg/eggs/logger"
@@ -73,6 +74,14 @@ func NewKTranslate(config *ktranslate.Config, log logger.ContextL, registry go_m
 		logTee:      logTee,
 		msgsc:       make(chan *kt.Output, 60),
 		tooBig:      make(chan int, CHAN_SLACK),
+		shutdown:    shutdown,
+	}
+
+	// If there's a config manager, start this now.
+	if confMgr, err := cfgMngr.NewConfig(cfgMngr.ConfigProvider(config.CfgManager.ConfigImpl), log.GetLogger().GetUnderlyingLogger(), config); err != nil {
+		return nil, err
+	} else {
+		kc.confMgr = confMgr
 	}
 
 	if v := config.API.DeviceFile; v != "" {
@@ -197,16 +206,6 @@ func NewKTranslate(config *ktranslate.Config, log logger.ContextL, registry go_m
 	// Get some randomness
 	rand.Seed(time.Now().UnixNano())
 
-	// If we are monitoring the conf file, kick this off now.
-	if config.Server.MonitorConf {
-		go func() {
-			err := kc.monitorConf(config.Server, shutdown)
-			if err != nil {
-				kc.log.Errorf("Cannot monitor config: %v", err)
-			}
-		}()
-	}
-
 	return kc, nil
 }
 
@@ -234,6 +233,9 @@ func (kc *KTranslate) cleanup() {
 	if kc.syslog != nil {
 		kc.syslog.Close()
 	}
+	if kc.confMgr != nil {
+		kc.confMgr.Close()
+	}
 }
 
 // GetStatus implements the baseserver.Service interface.
@@ -569,6 +571,10 @@ func (kc *KTranslate) listenHTTP() {
 func (kc *KTranslate) Run(ctx context.Context) error {
 	defer kc.cleanup()
 
+	if kc.confMgr != nil {
+		go kc.confMgr.Run(ctx, kc.newConfig)
+	}
+
 	format := formats.Format(kc.config.Format)
 	formatRollup := formats.Format(kc.config.FormatRollup)
 	compression := kt.Compression(kc.config.Compression)
@@ -646,12 +652,12 @@ 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.SNMPInput.Enable {
 		if kc.config.EnableSNMPDiscovery { // Here, we're just returning the list of devices on the network which might speak snmp.
-			_, err := snmp.Discover(ctx, kc.log, 0, kc.config.SNMPInput, kc.apic)
+			_, err := snmp.Discover(ctx, kc.log, 0, kc.config.SNMPInput, kc.apic, kc.confMgr)
 			return err
 		}
 		assureInput()
 		kc.metrics.SnmpDeviceData = kt.NewSnmpMetricSet(kc.registry)
-		err := snmp.StartSNMPPolls(ctx, kc.inputChan, kc.metrics.SnmpDeviceData, kc.registry, kc.apic, kc.log, kc.config.SNMPInput, kc.resolver)
+		err := snmp.StartSNMPPolls(ctx, kc.inputChan, kc.metrics.SnmpDeviceData, kc.registry, kc.apic, kc.log, kc.config.SNMPInput, kc.resolver, kc.confMgr)
 		if err != nil {
 			return err
 		}
diff --git a/pkg/cat/mon.go b/pkg/cat/mon.go
index d9d47ec..9bece25 100644
--- a/pkg/cat/mon.go
+++ b/pkg/cat/mon.go
@@ -2,49 +2,11 @@ package cat
 
 import (
 	"github.com/kentik/ktranslate"
-
-	"github.com/fsnotify/fsnotify"
 )
 
-func (kc *KTranslate) monitorConf(cfg *ktranslate.ServerConfig, shutdown func(string)) error {
-	kc.log.Infof("Monitoring %s for changes", cfg.CfgPath)
-
-	// Create new watcher.
-	watcher, err := fsnotify.NewWatcher()
-	if err != nil {
-		return err
-	}
-	defer watcher.Close()
-
-	// Start listening for events.
-	go func() {
-		for {
-			select {
-			case event, ok := <-watcher.Events:
-				if !ok {
-					return
-				}
-				if event.Has(fsnotify.Write) {
-					kc.log.Warnf("Write detected on %s, shutting down", cfg.CfgPath)
-					shutdown("Config file changed")
-				}
-			case err, ok := <-watcher.Errors:
-				if !ok {
-					return
-				}
-				kc.log.Infof("error:", err)
-			}
-		}
-	}()
-
-	// Add a path.
-	err = watcher.Add(cfg.CfgPath)
-	if err != nil {
-		return err
-	}
-
-	// Block forever.
-	<-make(chan struct{})
-
+// Callback for when theres a config managment service which detects a change.
+func (kc *KTranslate) newConfig(newC *ktranslate.Config) error {
+	kc.log.Warnf("Write detected on %s, shutting down", newC.Server.CfgPath)
+	kc.shutdown("Config file changed")
 	return nil
 }
diff --git a/pkg/cat/types.go b/pkg/cat/types.go
index 6eac234..d3feb8e 100644
--- a/pkg/cat/types.go
+++ b/pkg/cat/types.go
@@ -9,6 +9,7 @@ import (
 
 	"github.com/kentik/ktranslate/pkg/api"
 	"github.com/kentik/ktranslate/pkg/cat/auth"
+	"github.com/kentik/ktranslate/pkg/config"
 	"github.com/kentik/ktranslate/pkg/filter"
 	"github.com/kentik/ktranslate/pkg/formats"
 	"github.com/kentik/ktranslate/pkg/inputs/flow"
@@ -78,6 +79,8 @@ type KTranslate struct {
 	enricher     *enrich.Enricher
 	logTee       chan string
 	authConfig   *auth.AuthConfig
+	confMgr      config.ConfigManager
+	shutdown     func(string)
 }
 
 type CustomMapper struct {
diff --git a/pkg/config/config.go b/pkg/config/config.go
new file mode 100644
index 0000000..50f4d89
--- /dev/null
+++ b/pkg/config/config.go
@@ -0,0 +1,52 @@
+package config
+
+/**
+Interface to manage configs.
+*/
+
+import (
+	"context"
+	"flag"
+	"fmt"
+
+	"github.com/kentik/ktranslate"
+	"github.com/kentik/ktranslate/pkg/config/local"
+	"github.com/kentik/ktranslate/pkg/config/nr"
+	"github.com/kentik/ktranslate/pkg/eggs/logger"
+	"github.com/kentik/ktranslate/pkg/kt"
+)
+
+type ConfigManager interface {
+	Run(context.Context, func(*ktranslate.Config) error) // Run takes a context and a callback function to call whenever there is a new update to process
+	DeviceDiscovery(kt.DeviceMap)                        // called whenever there is a new snmp device discovery to parse.
+	Close()                                              // Called on shutdown of ktrans.
+}
+
+type ConfigProvider string
+
+const (
+	NewRelicConfig ConfigProvider = "new_relic"
+	LocalConfig    ConfigProvider = "local"
+	NoConfig       ConfigProvider = ""
+)
+
+var (
+	configProvider string
+)
+
+func init() {
+	flag.StringVar(&configProvider, "config_provider", "", "Implementation of which provider controls the config process. Can be one of (new_relic,local)")
+}
+
+func NewConfig(prov ConfigProvider, log logger.Underlying, config *ktranslate.Config) (ConfigManager, error) {
+	switch prov {
+	case NewRelicConfig:
+		return nr.NewConfig(log, config)
+	case LocalConfig:
+		return local.NewConfig(log, config)
+	case NoConfig:
+		return nil, nil
+	default:
+		return nil, fmt.Errorf("Unknown config provider %v", prov)
+	}
+}
diff --git a/pkg/config/local/local.go b/pkg/config/local/local.go
new file mode 100644
index 0000000..b62b141
--- /dev/null
+++ b/pkg/config/local/local.go
@@ -0,0 +1,76 @@
+package local
+
+import (
+	"context"
+
+	"github.com/kentik/ktranslate"
+	"github.com/kentik/ktranslate/pkg/eggs/logger"
+	"github.com/kentik/ktranslate/pkg/kt"
+
+	"github.com/fsnotify/fsnotify"
+)
+
+type LocalConfig struct {
+	logger.ContextL
+	currentConfig *ktranslate.Config
+	watcher       *fsnotify.Watcher
+}
+
+func NewConfig(log logger.Underlying, cfg *ktranslate.Config) (*LocalConfig, error) {
+	lc := LocalConfig{
+		ContextL:      logger.NewContextLFromUnderlying(logger.SContext{S: "localConfig"}, log),
+		currentConfig: cfg,
+	}
+
+	lc.Infof("Monitoring %s for changes", lc.currentConfig.Server.CfgPath)
+
+	// Create new watcher.
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		return nil, err
+	}
+
+	// Add a path.
+	err = watcher.Add(lc.currentConfig.Server.CfgPath)
+	if err != nil {
+		return nil, err
+	}
+
+	lc.watcher = watcher
+
+	return &lc, nil
+}
+
+func (lc *LocalConfig) Run(ctx context.Context, cb func(*ktranslate.Config) error) {
+	lc.Infof("config checker running")
+	for {
+		select {
+		case event, ok := <-lc.watcher.Events:
+			if !ok {
+				return
+			}
+			if event.Has(fsnotify.Write) {
+				err := cb(lc.currentConfig)
+				if err != nil {
+					lc.Errorf("Cannot update to new config: %v", err)
+				}
+			}
+		case err, ok := <-lc.watcher.Errors:
+			if !ok {
+				return
+			}
+			lc.Infof("error:", err)
+		case <-ctx.Done():
+			lc.Infof("config checker done")
+			return
+		}
+	}
+}
+
+func (lc *LocalConfig) DeviceDiscovery(devices kt.DeviceMap) {
+	// NOOP
+}
+
+func (lc *LocalConfig) Close() {
+	lc.watcher.Close()
+}
diff --git a/pkg/config/nr/nr.go b/pkg/config/nr/nr.go
new file mode 100644
index 0000000..372300d
--- /dev/null
+++ b/pkg/config/nr/nr.go
@@ -0,0 +1,64 @@
+package nr
+
+import (
+	"context"
+	"time"
+
+	"github.com/kentik/ktranslate"
+	"github.com/kentik/ktranslate/pkg/eggs/logger"
+	"github.com/kentik/ktranslate/pkg/kt"
+)
+
+type NRConfig struct {
+	logger.ContextL
+	currentConfig *ktranslate.Config
+}
+
+func NewConfig(log logger.Underlying, cfg *ktranslate.Config) (*NRConfig, error) {
+	nr := NRConfig{
+		ContextL:      logger.NewContextLFromUnderlying(logger.SContext{S: "nrConfig"}, log),
+		currentConfig: cfg,
+	}
+
+	return &nr, nil
+}
+
+func (nr *NRConfig) Run(ctx context.Context, cb func(*ktranslate.Config) error) {
+	checkTicker := time.NewTicker(time.Second * time.Duration(nr.currentConfig.CfgManager.PollTimeSec))
+	defer checkTicker.Stop()
+
+	nr.Infof("config checker running")
+	for {
+		select {
+		case <-checkTicker.C:
+			// Get config
+			newConfig, newVersion, err := nr.getConfig()
+			if err != nil {
+				nr.Errorf("Cannot load new config: %v", err)
+			}
+
+			if newConfig != nil && newVersion {
+				nr.currentConfig = newConfig
+				err := cb(newConfig)
+				if err != nil {
+					nr.Errorf("Cannot update to new config: %v", err)
+				}
+			}
+		case <-ctx.Done():
+			nr.Infof("config checker done")
+			return
+		}
+	}
+}
+
+func (nr *NRConfig) DeviceDiscovery(devices kt.DeviceMap) {
+
+}
+
+func (nr *NRConfig) Close() {
+
+}
+
+func (nr *NRConfig) getConfig() (*ktranslate.Config, bool, error) {
+	return nil, false, nil
+}
diff --git a/pkg/formats/ddog/ddog.go b/pkg/formats/ddog/ddog.go
new file mode 100644
index 0000000..d3b10c4
--- /dev/null
+++ b/pkg/formats/ddog/ddog.go
@@ -0,0 +1,483 @@
+package ddog
+
+import (
+	"bytes"
+	"compress/gzip"
+	"fmt"
+	"strconv"
+	"strings"
+	"sync"
+
+	"github.com/kentik/ktranslate/pkg/eggs/logger"
+	"github.com/kentik/ktranslate/pkg/formats/util"
+	"github.com/kentik/ktranslate/pkg/kt"
+	"github.com/kentik/ktranslate/pkg/rollup"
+
+	"github.com/DataDog/datadog-api-client-go/v2/api/datadog"
+	"github.com/DataDog/datadog-api-client-go/v2/api/datadogV2"
+)
+
+type DDogFormat struct {
+	logger.ContextL
+	compression  kt.Compression
+	doGz         bool
+	lastMetadata map[string]*kt.LastMetadata
+	invalids     map[string]bool
+	mux          sync.RWMutex
+
+	EventChan chan []byte
+}
+
+func NewFormat(log logger.Underlying, compression kt.Compression) (*DDogFormat, error) {
+	jf := &DDogFormat{
+		compression:  compression,
+		ContextL:     logger.NewContextLFromUnderlying(logger.SContext{S: "ddogFormat"}, log),
+		doGz:         false,
+		invalids:     map[string]bool{},
+		lastMetadata: map[string]*kt.LastMetadata{},
+		EventChan:    make(chan []byte, 100), // Used for sending events to the event API.
+	}
+
+	switch compression {
+	case kt.CompressionNone:
+		jf.doGz = false
+	case kt.CompressionGzip:
+		jf.doGz = true
+	default:
+		return nil, fmt.Errorf("Invalid compression (%s): format ddog only supports none|gzip", compression)
+	}
+
+	return jf, nil
+}
+
+func (f *DDogFormat) To(msgs []*kt.JCHF, serBuf []byte) (*kt.Output, error) {
+	ms := datadogV2.NewMetricPayloadWithDefaults()
+	for _, m := range msgs {
+		err := f.toDDogMetric(m, ms)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if len(ms.Series) == 0 {
+		return nil, nil
+	}
+
+	target, err := json.Marshal(ms)
+	if err != nil {
+		return nil, err
+	}
+
+	if !f.doGz {
+		return kt.NewOutput(target), nil
+	}
+
+	buf := bytes.NewBuffer(serBuf)
+	buf.Reset()
+	zw, err := gzip.NewWriterLevel(buf, gzip.DefaultCompression)
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = zw.Write(target)
+	if err != nil {
+		return nil, err
+	}
+
+	err = zw.Close()
+	if err != nil {
+		return nil, err
+	}
+
+	return kt.NewOutput(buf.Bytes()), nil
+}
+
+func (f *DDogFormat) From(raw *kt.Output) ([]map[string]interface{}, error) {
+	values := make([]map[string]interface{}, 0)
+	return values, nil
+}
+
+func (f *DDogFormat) Rollup(rolls []rollup.Rollup) (*kt.Output, error) {
+	return nil, nil
+}
+
+func (f *DDogFormat) toDDogMetric(in *kt.JCHF, ms *datadogV2.MetricPayload) error {
+	switch in.EventType {
+	case kt.KENTIK_EVENT_SNMP_DEV_METRIC:
+		return f.fromSnmpDeviceMetric(in, ms)
+	case kt.KENTIK_EVENT_SNMP_INT_METRIC:
+		return f.fromSnmpInterfaceMetric(in, ms)
+	case kt.KENTIK_EVENT_SNMP_METADATA:
+		return f.fromSnmpMetadata(in, ms)
+	case kt.KENTIK_EVENT_KTRANS_METRIC:
+		return f.fromKtranslate(in, ms)
+	case kt.KENTIK_EVENT_TYPE:
+		return f.fromKflow(in, ms)
+	case kt.KENTIK_EVENT_SYNTH:
+		return f.fromKSynth(in, ms)
+	}
+
+	return nil
+}
+
+func (f *DDogFormat) fromKSynth(in *kt.JCHF, ms *datadogV2.MetricPayload) error {
+	metrics := util.GetSynMetricNameSet(in.CustomInt["result_type"])
+	attr := map[string]interface{}{}
+	f.mux.RLock()
+	util.SetAttr(attr, in, metrics, f.lastMetadata[in.DeviceName], true)
+	f.mux.RUnlock()
+	tags := getDDMetricTags(attr)
+
+	for m, name := range metrics {
+		switch name.Name {
+		case "avg_rtt", "jit_rtt", "time", "code", "port", "status", "ttlb", "sent", "lost":
+			ms.Series = append(ms.Series, datadogV2.MetricSeries{
+				Metric: "kentik.synth." + m,
+				Type:   datadogV2.METRICINTAKETYPE_GAUGE.Ptr(),
+				Points: []datadogV2.MetricPoint{
+					{
+						Timestamp: datadog.PtrInt64(in.Timestamp),
+						Value:     datadog.PtrFloat64(float64(in.CustomInt[m])),
+					},
+				},
+				Tags: tags,
+			})
+		}
+	}
+
+	return nil
+}
+
+func (f *DDogFormat) fromKflow(in *kt.JCHF, ms *datadogV2.MetricPayload) error {
+	// Map the basic strings into here.
+	attr := map[string]interface{}{}
+	metrics := map[string]kt.MetricInfo{"in_bytes": {}, "out_bytes": {}, "in_pkts": {}, "out_pkts": {}, "latency_ms": {}}
+	f.mux.RLock()
+	util.SetAttr(attr, in, metrics, f.lastMetadata[in.DeviceName], true)
+	f.mux.RUnlock()
+	tags := getDDMetricTags(attr)
+
+	for m := range metrics {
+		switch m {
+		case "in_bytes":
+			ms.Series = append(ms.Series, datadogV2.MetricSeries{
+				Metric: "kentik.flow." + m,
+				Type:   datadogV2.METRICINTAKETYPE_GAUGE.Ptr(),
+				Points: []datadogV2.MetricPoint{
+					{
+						Timestamp: datadog.PtrInt64(in.Timestamp),
+						Value:     datadog.PtrFloat64(float64(int64(in.InBytes * uint64(in.SampleRate)))),
+					},
+				},
+				Tags: tags,
+			})
+		case "out_bytes":
+			ms.Series = append(ms.Series, datadogV2.MetricSeries{
+				Metric: "kentik.flow." + m,
+				Type:   datadogV2.METRICINTAKETYPE_GAUGE.Ptr(),
+				Points: []datadogV2.MetricPoint{
+					{
+						Timestamp: datadog.PtrInt64(in.Timestamp),
+						Value:     datadog.PtrFloat64(float64(int64(in.OutBytes * uint64(in.SampleRate)))),
+					},
+				},
+				Tags: tags,
+			})
+		case "in_pkts":
+			ms.Series = append(ms.Series, datadogV2.MetricSeries{
+				Metric: "kentik.flow." + m,
+				Type:   datadogV2.METRICINTAKETYPE_GAUGE.Ptr(),
+				Points: []datadogV2.MetricPoint{
+					{
+						Timestamp: datadog.PtrInt64(in.Timestamp),
+						Value:     datadog.PtrFloat64(float64(int64(in.InPkts * uint64(in.SampleRate)))),
+					},
+				},
+				Tags: tags,
+			})
+		case "out_pkts":
+			ms.Series = append(ms.Series, datadogV2.MetricSeries{
+				Metric: "kentik.flow." + m,
+				Type:   datadogV2.METRICINTAKETYPE_GAUGE.Ptr(),
+				Points: []datadogV2.MetricPoint{
+					{
+						Timestamp: datadog.PtrInt64(in.Timestamp),
+						Value:     datadog.PtrFloat64(float64(int64(in.OutPkts * uint64(in.SampleRate)))),
+					},
+				},
+				Tags: tags,
+			})
+		case "latency_ms":
+			ms.Series = append(ms.Series, datadogV2.MetricSeries{
+				Metric: "kentik.flow." + m,
+				Type:   datadogV2.METRICINTAKETYPE_GAUGE.Ptr(),
+				Points: []datadogV2.MetricPoint{
+					{
+						Timestamp: datadog.PtrInt64(in.Timestamp),
+						Value:     datadog.PtrFloat64(float64(in.CustomInt["appl_latency_ms"])),
+					},
+				},
+				Tags: tags,
+			})
+		}
+	}
+
+	return nil
+}
+
+func (f *DDogFormat) fromKtranslate(in *kt.JCHF, ms *datadogV2.MetricPayload) error {
+	// Map the basic strings into here.
+	attr := map[string]interface{}{}
+	metrics := map[string]kt.MetricInfo{"name": kt.MetricInfo{}, "value": kt.MetricInfo{}, "count": kt.MetricInfo{}, "one-minute": kt.MetricInfo{}, "95-percentile": kt.MetricInfo{}, "du": kt.MetricInfo{}}
+	f.mux.RLock()
+	util.SetAttr(attr, in, metrics, f.lastMetadata[in.DeviceName], false)
+	f.mux.RUnlock()
+	tags := getDDMetricTags(attr)
+	var value *float64 = nil
+
+	switch in.CustomStr["type"] {
+	case "counter":
+		if in.CustomStr["force"] == "true" || in.CustomBigInt["count"] > 0 {
+			value = datadog.PtrFloat64(float64(in.CustomBigInt["count"]) / 100)
+		}
+	case "gauge":
+		if in.CustomStr["force"] == "true" || in.CustomBigInt["value"] > 0 {
+			value = datadog.PtrFloat64(float64(in.CustomBigInt["value"]) / 100)
+		}
+	case "histogram":
+		if in.CustomStr["force"] == "true" || in.CustomBigInt["95-percentile"] > 0 {
+			value = datadog.PtrFloat64(float64(in.CustomBigInt["95-percentile"]) / 100)
+		}
+	case "meter":
+		if in.CustomStr["force"] == "true" || in.CustomBigInt["one-minute"] > 0 {
+			value = datadog.PtrFloat64(float64(in.CustomBigInt["one-minute"]) / 100)
+		}
+	case "timer":
+		if in.CustomStr["force"] == "true" || in.CustomBigInt["95-percentile"] > 0 {
+			value = datadog.PtrFloat64(float64(in.CustomBigInt["95-percentile"]) / 100)
+		}
+	}
+
+	if value != nil {
+		ms.Series = append(ms.Series, datadogV2.MetricSeries{
+			Metric: in.CustomStr["name"],
+			Type:   datadogV2.METRICINTAKETYPE_GAUGE.Ptr(),
+			Points: []datadogV2.MetricPoint{
+				{
+					Timestamp: datadog.PtrInt64(in.Timestamp),
+					Value:     value,
+				},
+			},
+			Tags: tags,
+		})
+	}
+
+	return nil
+}
+
+func (f *DDogFormat) fromSnmpMetadata(in *kt.JCHF, ms *datadogV2.MetricPayload) error {
+	if in.DeviceName == "" { // Only run if this is set.
+		return nil
+	}
+
+	lm := util.SetMetadata(in)
+
+	f.mux.Lock()
+	defer f.mux.Unlock()
+	f.lastMetadata[in.DeviceName] = lm
+
+	return nil
+}
+
+func (f *DDogFormat) fromSnmpDeviceMetric(in *kt.JCHF, ms *datadogV2.MetricPayload) error {
+	metrics := in.CustomMetrics
+	attr := map[string]interface{}{}
+	f.mux.RLock()
+	util.SetAttr(attr, in, metrics, f.lastMetadata[in.DeviceName], true)
+	f.mux.RUnlock()
+
+	for m, name := range metrics {
+		if m == "" {
+			f.Errorf("Missing metric name, skipping %v", attr)
+			continue
+		}
+		if _, ok := in.CustomBigInt[m]; ok {
+			attrNew := util.CopyAttrForSnmp(attr, m, name, f.lastMetadata[in.DeviceName], true, false)
+			if util.DropOnFilter(attrNew, f.lastMetadata[in.DeviceName], false) {
+				continue // This Metric isn't in the white list so lets drop it.
+			}
+			seriesName := "kentik.snmp." + m
+			tags := getDDMetricTags(attrNew)
+			if name.Format == kt.FloatMS {
+				ms.Series = append(ms.Series, datadogV2.MetricSeries{
+					Metric: seriesName,
+					Type:   datadogV2.METRICINTAKETYPE_GAUGE.Ptr(),
+					Points: []datadogV2.MetricPoint{
+						{
+							Timestamp: datadog.PtrInt64(in.Timestamp),
+							Value:     datadog.PtrFloat64(float64(float64(in.CustomBigInt[m]) / 1000)),
+						},
+					},
+					Tags: tags,
+				})
+			} else {
+				ms.Series = append(ms.Series, datadogV2.MetricSeries{
+					Metric: seriesName,
+					Type:   datadogV2.METRICINTAKETYPE_GAUGE.Ptr(),
+					Points: []datadogV2.MetricPoint{
+						{
+							Timestamp: datadog.PtrInt64(in.Timestamp),
+							Value:     datadog.PtrFloat64(float64(in.CustomBigInt[m])),
+						},
+					},
+					Tags: tags,
+				})
+			}
+		}
+	}
+
+	return nil
+}
+
+func (f *DDogFormat) fromSnmpInterfaceMetric(in *kt.JCHF, ms *datadogV2.MetricPayload) error {
+	metrics := in.CustomMetrics
+	attr := map[string]interface{}{}
+	f.mux.RLock()
+	defer f.mux.RUnlock()
+	util.SetAttr(attr, in, metrics, f.lastMetadata[in.DeviceName], true)
+
+	profileName := "snmp"
+	for m, name := range metrics {
+		if m == "" {
+			f.Errorf("Missing metric name, skipping %v", attr)
+			continue
+		}
+		if _, ok := in.CustomBigInt[m]; ok {
+			attrNew := util.CopyAttrForSnmp(attr, m, name, f.lastMetadata[in.DeviceName], true, false)
+			if util.DropOnFilter(attrNew, f.lastMetadata[in.DeviceName], true) {
+				continue // This Metric isn't in the white list so lets drop it.
+			}
+
+			profileName = name.Profile
+			seriesName := "kentik.snmp." + m
+			tags := getDDMetricTags(attrNew)
+			if name.Format == kt.FloatMS {
+				ms.Series = append(ms.Series, datadogV2.MetricSeries{
+					Metric: seriesName,
+					Type:   datadogV2.METRICINTAKETYPE_GAUGE.Ptr(),
+					Points: []datadogV2.MetricPoint{
+						{
+							Timestamp: datadog.PtrInt64(in.Timestamp),
+							Value:     datadog.PtrFloat64(float64(float64(in.CustomBigInt[m]) / 1000)),
+						},
+					},
+					Tags: tags,
+				})
+			} else {
+				ms.Series = append(ms.Series, datadogV2.MetricSeries{
+					Metric: seriesName,
+					Type:   datadogV2.METRICINTAKETYPE_GAUGE.Ptr(),
+					Points: []datadogV2.MetricPoint{
+						{
+							Timestamp: datadog.PtrInt64(in.Timestamp),
+							Value:     datadog.PtrFloat64(float64(in.CustomBigInt[m])),
+						},
+					},
+					Tags: tags,
+				})
+			}
+		}
+	}
+
+	// grab rates computed over time here if possible.
+	if f.lastMetadata[in.DeviceName] != nil {
+		f.setRates(ms, "In", in, attr, profileName)
+		f.setRates(ms, "Out", in, attr, profileName)
+	}
+
+	return nil
+}
+
+func (f *DDogFormat) setRates(ms *datadogV2.MetricPayload, direction string, in *kt.JCHF, attr map[string]interface{}, profileName string) {
+	var port kt.IfaceID
+	if direction == "In" {
+		port = in.InputPort
+	} else {
+		port = in.OutputPort
+	}
+	utilName := fmt.Sprintf("If%sUtilization", direction)
+	bitRate := fmt.Sprintf("If%sBitRate", direction)
+	pktRate := fmt.Sprintf("If%sPktRate", direction)
+	totalBytes := in.CustomBigInt[fmt.Sprintf("ifHC%sOctets", direction)]
+	totalPkts := in.CustomBigInt[fmt.Sprintf("ifHC%sUcastPkts", direction)] + in.CustomBigInt[fmt.Sprintf("ifHC%sMulticastPkts", direction)] + in.CustomBigInt[fmt.Sprintf("ifHC%sBroadcastPkts", direction)]
+
+	if ii, ok := f.lastMetadata[in.DeviceName].InterfaceInfo[port]; ok {
+		if speed, ok := ii["Speed"]; ok {
+			if ispeed, ok := speed.(int32); ok {
+				uptime := in.CustomBigInt["Uptime"]
+				uptimeSpeed := 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 := util.CopyAttrForSnmp(attr, utilName, kt.MetricInfo{Oid: "computed", Mib: "computed", Profile: profileName, Table: "if"}, f.lastMetadata[in.DeviceName], true, false)
+					if !util.DropOnFilter(attrNew, f.lastMetadata[in.DeviceName], true) {
+						tags := getDDMetricTags(attrNew)
+						if totalBytes > 0 {
+							ms.Series = append(ms.Series, datadogV2.MetricSeries{
+								Metric: "kentik.snmp." + utilName,
+								Type:   datadogV2.METRICINTAKETYPE_GAUGE.Ptr(),
+								Points: []datadogV2.MetricPoint{
+									{
+										Timestamp: datadog.PtrInt64(in.Timestamp),
+										Value:     datadog.PtrFloat64(float64(totalBytes*8*100) / float64(uptimeSpeed)),
+									},
+								},
+								Tags: tags,
+							},
+								datadogV2.MetricSeries{
+									Metric: "kentik.snmp." + bitRate,
+									Type:   datadogV2.METRICINTAKETYPE_GAUGE.Ptr(),
+									Points: []datadogV2.MetricPoint{
+										{
+											Timestamp: datadog.PtrInt64(in.Timestamp),
+											Value:     datadog.PtrFloat64(float64(totalBytes*8*100) / float64(uptime)),
+										},
+									},
+									Tags: tags,
+								},
+							)
+						}
+						if totalPkts > 0 {
+							ms.Series = append(ms.Series, datadogV2.MetricSeries{
+								Metric: "kentik.snmp." + pktRate,
+								Type:   datadogV2.METRICINTAKETYPE_GAUGE.Ptr(),
+								Points: []datadogV2.MetricPoint{
+									{
+										Timestamp: datadog.PtrInt64(in.Timestamp),
+										Value:     datadog.PtrFloat64(float64(totalPkts*100) / float64(uptime)),
+									},
+								},
+								Tags: tags,
+							})
+						}
+					}
+				}
+			}
+		}
+	}
+}
+
+func getDDMetricTags(attrNew map[string]interface{}) []string {
+	tags := []string{}
+	for k, v := range attrNew {
+		switch nv := v.(type) {
+		case string:
+			tags = append(tags, strings.Join([]string{k, nv}, ":"))
+		case int64:
+			tags = append(tags, strings.Join([]string{k, strconv.Itoa(int(nv))}, ":"))
+		case int32:
+			tags = append(tags, strings.Join([]string{k, strconv.Itoa(int(nv))}, ":"))
+		case kt.Cid:
+			tags = append(tags, strings.Join([]string{k, strconv.Itoa(int(nv))}, ":"))
+		}
+	}
+	return tags
+}
diff --git a/pkg/formats/ddog/init.go b/pkg/formats/ddog/init.go
new file mode 100644
index 0000000..de08e0f
--- /dev/null
+++ b/pkg/formats/ddog/init.go
@@ -0,0 +1,16 @@
+package ddog
+
+import (
+	"github.com/json-iterator/go"
+)
+
+// fast JSON encoding
+var json = jsoniter.ConfigFastest
+
+// jsonSorted is fast, but still sorts keys
+var jsonSorted = jsoniter.Config{
+	IndentionStep:          4,
+	EscapeHTML:             false,
+	SortMapKeys:            true,
+	ValidateJsonRawMessage: false,
+}.Froze()
diff --git a/pkg/formats/format.go b/pkg/formats/format.go
index 4a273d3..fafbbfe 100644
--- a/pkg/formats/format.go
+++ b/pkg/formats/format.go
@@ -8,6 +8,7 @@ import (
 
 	"github.com/kentik/ktranslate/pkg/formats/avro"
 	"github.com/kentik/ktranslate/pkg/formats/carbon"
+	"github.com/kentik/ktranslate/pkg/formats/ddog"
 	"github.com/kentik/ktranslate/pkg/formats/elasticsearch"
 	"github.com/kentik/ktranslate/pkg/formats/influx"
 	"github.com/kentik/ktranslate/pkg/formats/json"
@@ -42,6 +43,7 @@ const (
 	FORMAT_NR                   = "new_relic"
 	FORMAT_NRM                  = "new_relic_metric"
 	FORMAT_SPLUNK               = "splunk"
+	FORMAT_DDOG                 = "ddog"
 	FORMAT_KFLOW                = "kflow"
 )
 
@@ -65,6 +67,8 @@ func NewFormat(format Format, log logger.Underlying, registry go_metrics.Registr
 		return json.NewFormat(log, compression, true)
 	case FORMAT_NRM:
 		return nrm.NewFormat(log, compression)
+	case FORMAT_DDOG:
+		return ddog.NewFormat(log, compression)
 	case FORMAT_SPLUNK:
 		return splunk.NewFormat(log, compression)
 	case FORMAT_KFLOW:
diff --git a/pkg/formats/influx/influx.go b/pkg/formats/influx/influx.go
index c84724c..12dd2de 100644
--- a/pkg/formats/influx/influx.go
+++ b/pkg/formats/influx/influx.go
@@ -21,11 +21,12 @@ import (
 
 var (
 	measurementPrefix string
+	namespaceTokenSep string
 )
 
 func init() {
 	flag.StringVar(&measurementPrefix, "influxdb_measurement_prefix", "", "Prefix metric names with this")
-
+	flag.StringVar(&namespaceTokenSep, "influxdb_namespace_token", ":", "Use this token to seperate namespaces")
 }
 
 type InfluxFormat struct {
@@ -104,6 +105,10 @@ func NewMergedInfluxData(s InfluxDataSet) *InfluxData {
 
 type InfluxDataSet []InfluxData
 
+func (f InfluxDataSet) Len() int           { return len(f) }
+func (f InfluxDataSet) Swap(i, j int)      { f[i], f[j] = f[j], f[i] }
+func (f InfluxDataSet) Less(i, j int) bool { return f[i].Name < f[j].Name }
+
 func (f *InfluxFormat) report(err error, format string, args ...interface{}) {
 	str := fmt.Sprintf(format, args...)
 	f.invalidsMux.RLock()
@@ -123,7 +128,10 @@ func (f *InfluxFormat) report(err error, format string, args ...interface{}) {
 }
 
 func (f *InfluxFormat) Bytes(s InfluxDataSet) []byte {
-	// First map common prefixes.
+	// Sort the output before anything.
+	sort.Sort(s)
+
+	// Next map common prefixes.
 	prefixes := map[string][]InfluxData{}
 	for _, l := range s {
 		prefix := l.Prefix()
@@ -262,6 +270,8 @@ func NewFormat(log logger.Underlying, registry go_metrics.Registry, compression
 		config: cfg,
 	}
 
+	namespaceTokenSep = cfg.NamespaceToken
+
 	return jf, nil
 }
 
@@ -519,17 +529,22 @@ func (f *InfluxFormat) fromSnmpDeviceMetric(in *kt.JCHF) []InfluxData {
 			}
 
 			mib := getMib(attrNew, ip)
+			metricName := m
+			if strings.Contains(metricName, namespaceTokenSep) {
+				pts := strings.SplitN(metricName, namespaceTokenSep, 2)
+				metricName = pts[1]
+			}
 			if name.Format == kt.FloatMS {
 				results = append(results, InfluxData{
 					Name:        f.config.MeasurementPrefix + mib,
-					FieldsFloat: map[string]float64{m: float64(float64(in.CustomBigInt[m]) / 1000)},
+					FieldsFloat: map[string]float64{metricName: float64(float64(in.CustomBigInt[m]) / 1000)},
 					Timestamp:   in.Timestamp,
 					Tags:        attrNew,
 				})
 			} else {
 				results = append(results, InfluxData{
 					Name:      f.config.MeasurementPrefix + mib,
-					Fields:    map[string]int64{m: int64(in.CustomBigInt[m])},
+					Fields:    map[string]int64{metricName: int64(in.CustomBigInt[m])},
 					Timestamp: in.Timestamp,
 					Tags:      attrNew,
 				})
@@ -563,17 +578,22 @@ func (f *InfluxFormat) fromSnmpInterfaceMetric(in *kt.JCHF) []InfluxData {
 			}
 
 			mib := getMib(attrNew, ip)
+			metricName := m
+			if strings.Contains(metricName, namespaceTokenSep) {
+				pts := strings.SplitN(metricName, namespaceTokenSep, 2)
+				metricName = pts[1]
+			}
 			if name.Format == kt.FloatMS {
 				results = append(results, InfluxData{
 					Name:        f.config.MeasurementPrefix + mib,
-					FieldsFloat: map[string]float64{m: float64(float64(in.CustomBigInt[m]) / 1000)},
+					FieldsFloat: map[string]float64{metricName: float64(float64(in.CustomBigInt[m]) / 1000)},
 					Timestamp:   in.Timestamp,
 					Tags:        attrNew,
 				})
 			} else {
 				results = append(results, InfluxData{
 					Name:      f.config.MeasurementPrefix + mib,
-					Fields:    map[string]int64{m: int64(in.CustomBigInt[m])},
+					Fields:    map[string]int64{metricName: int64(in.CustomBigInt[m])},
 					Timestamp: in.Timestamp,
 					Tags:      attrNew,
 				})
@@ -656,6 +676,13 @@ func getMib(attr map[string]interface{}, ip interface{}) string {
 			delete(attr, k)
 			attr["index"] = v
 		}
+
+		// If the attribute has a namespace prefix, drop here.
+		if strings.Contains(k, namespaceTokenSep) {
+			pts := strings.SplitN(k, namespaceTokenSep, 2)
+			delete(attr, k)
+			attr[pts[1]] = v
+		}
 	}
 	if ip != nil {
 		attr["device_ip"] = ip
@@ -674,7 +701,12 @@ func getMib(attr map[string]interface{}, ip interface{}) string {
 	// If there's a table, add this to the info.
 	mibTable, ok := attr["mib-table"].(string)
 	if ok {
-		mib = mib + "::" + mibTable
+		// If the MIB is normalized use "/" as separator
+		if strings.HasPrefix(mib, "/") || strings.HasSuffix(mib, "/"){
+			mib = strings.TrimRight(mib, "/") + "/" + strings.TrimLeft(mibTable, "/")
+		} else {
+			mib = mib + "::" + mibTable
+		}
 		delete(attr, "mib-table")
 	}
 
@@ -683,8 +715,9 @@ func getMib(attr map[string]interface{}, ip interface{}) string {
 
 var dropFields = map[string]bool{
 	//"device_name": true,
-	"objectIdentifier": true,
-	"eventType":        true,
-	"provider":         true,
-	"SysServices":      true,
+	"objectIdentifier":     true,
+	"eventType":            true,
+	"provider":             true,
+	"SysServices":          true,
+	"instrumentation.name": true,
 }
diff --git a/pkg/inputs/snmp/disco.go b/pkg/inputs/snmp/disco.go
index 6ba9362..ad19c45 100644
--- a/pkg/inputs/snmp/disco.go
+++ b/pkg/inputs/snmp/disco.go
@@ -17,6 +17,7 @@ import (
 
 	"github.com/kentik/ktranslate"
 	"github.com/kentik/ktranslate/pkg/api"
+	"github.com/kentik/ktranslate/pkg/config"
 	"github.com/kentik/ktranslate/pkg/eggs/logger"
 	"github.com/kentik/ktranslate/pkg/inputs/snmp/metadata"
 	"github.com/kentik/ktranslate/pkg/inputs/snmp/mibs"
@@ -35,7 +36,7 @@ type SnmpDiscoDeviceStat struct {
 	delta    int
 }
 
-func Discover(ctx context.Context, log logger.ContextL, pollDuration time.Duration, cfg *ktranslate.SNMPInputConfig, apic *api.KentikApi) (*SnmpDiscoDeviceStat, error) {
+func Discover(ctx context.Context, log logger.ContextL, pollDuration time.Duration, cfg *ktranslate.SNMPInputConfig, apic *api.KentikApi, confMgr config.ConfigManager) (*SnmpDiscoDeviceStat, error) {
 	// First, parse the config file and see what we're doing.
 	snmpFile := cfg.SNMPFile
 	log.Infof("SNMP Discovery, loading config from %s", snmpFile)
@@ -141,15 +142,20 @@ func Discover(ctx context.Context, log logger.ContextL, pollDuration time.Durati
 		}
 	}
 
+	// Phone home if we have one of these set up.
+	if confMgr != nil && (stats.added > 0 || stats.replaced > 0) {
+		confMgr.DeviceDiscovery(conf.Devices)
+	}
+
 	time.Sleep(2 * time.Second) // Give logs time to get sent back.
 
 	return stats, nil
 }
 
-func RunDiscoOnTimer(ctx context.Context, c chan os.Signal, log logger.ContextL, pollTimeMin int, checkNow bool, cfg *ktranslate.SNMPInputConfig, apic *api.KentikApi) {
+func RunDiscoOnTimer(ctx context.Context, c chan os.Signal, log logger.ContextL, pollTimeMin int, checkNow bool, cfg *ktranslate.SNMPInputConfig, apic *api.KentikApi, confMgr config.ConfigManager) {
 	pt := time.Duration(pollTimeMin) * time.Minute
 	check := func() {
-		stats, err := Discover(ctx, log, pt, cfg, apic)
+		stats, err := Discover(ctx, log, pt, cfg, apic, confMgr)
 		if err != nil {
 			log.Errorf("Discovery SNMP Error: %v", err)
 		} else {
diff --git a/pkg/inputs/snmp/metadata/device_metadata.go b/pkg/inputs/snmp/metadata/device_metadata.go
index 3d51b51..e3875d0 100644
--- a/pkg/inputs/snmp/metadata/device_metadata.go
+++ b/pkg/inputs/snmp/metadata/device_metadata.go
@@ -39,17 +39,21 @@ var (
 type DeviceMetadata struct {
 	log     logger.ContextL
 	mibs    map[string]*kt.Mib
+	gconf   *kt.SnmpGlobalConfig
 	conf    *kt.SnmpDeviceConfig
 	metrics *kt.SnmpDeviceMetric
 	missing map[string]bool
 	basic   map[string]string
 }
 
-func NewDeviceMetadata(deviceMetadataMibs map[string]*kt.Mib, conf *kt.SnmpDeviceConfig, metrics *kt.SnmpDeviceMetric, log logger.ContextL) *DeviceMetadata {
+func NewDeviceMetadata(deviceMetadataMibs map[string]*kt.Mib, gconf *kt.SnmpGlobalConfig, conf *kt.SnmpDeviceConfig, metrics *kt.SnmpDeviceMetric, log logger.ContextL) *DeviceMetadata {
 	mibs := map[string]*kt.Mib{}
 	basic := map[string]string{}
-	for el := SNMP_device_metadata_oids.Front(); el != nil; el = el.Next() {
-		basic[el.Key.(string)] = el.Value.(string)
+
+	if !gconf.NoDeviceHardcodedOids {
+		for el := SNMP_device_metadata_oids.Front(); el != nil; el = el.Next() {
+			basic[el.Key.(string)] = el.Value.(string)
+		}
 	}
 
 	if len(deviceMetadataMibs) > 0 {
@@ -70,6 +74,7 @@ func NewDeviceMetadata(deviceMetadataMibs map[string]*kt.Mib, conf *kt.SnmpDevic
 		log:     log,
 		mibs:    mibs,
 		conf:    conf,
+		gconf:   gconf,
 		basic:   basic,
 		metrics: metrics,
 		missing: map[string]bool{},
diff --git a/pkg/inputs/snmp/metadata/interface_metadata.go b/pkg/inputs/snmp/metadata/interface_metadata.go
index 35446b3..6ddddb7 100644
--- a/pkg/inputs/snmp/metadata/interface_metadata.go
+++ b/pkg/inputs/snmp/metadata/interface_metadata.go
@@ -65,6 +65,7 @@ func NewInterfaceMetadata(interfaceMetadataMibs map[string]*kt.Mib, log logger.C
 		m.Set(el.Key, el.Value)
 	}
 
+
 	if len(interfaceMetadataMibs) > 0 {
 		oids := getFromCustomMap(interfaceMetadataMibs)
 		for _, oid := range oids {
diff --git a/pkg/inputs/snmp/metadata/poll.go b/pkg/inputs/snmp/metadata/poll.go
index 682fd5e..12d3caa 100644
--- a/pkg/inputs/snmp/metadata/poll.go
+++ b/pkg/inputs/snmp/metadata/poll.go
@@ -94,7 +94,7 @@ func NewPoller(server *gosnmp.GoSNMP, gconf *kt.SnmpGlobalConfig, conf *kt.SnmpD
 		server:                server,
 		interval:              DEFAULT_INTERVAL,
 		interfaceMetadata:     NewInterfaceMetadata(interfaceMetadataMibs, log),
-		deviceMetadata:        NewDeviceMetadata(deviceMetadataMibs, conf, metrics, log),
+		deviceMetadata:        NewDeviceMetadata(deviceMetadataMibs, gconf, conf, metrics, log),
 		jchfChan:              jchfChan,
 		metrics:               metrics,
 		deviceMetadataMibs:    deviceMetadataMibs,
diff --git a/pkg/inputs/snmp/metrics/device_metrics.go b/pkg/inputs/snmp/metrics/device_metrics.go
index 937512b..2de0307 100644
--- a/pkg/inputs/snmp/metrics/device_metrics.go
+++ b/pkg/inputs/snmp/metrics/device_metrics.go
@@ -205,6 +205,11 @@ func (dm *DeviceMetrics) pollFromConfig(ctx context.Context, server *gosnmp.GoSN
 			}
 			dmr.customBigInt[oidName] = snmp_util.ToInt64(wrapper.variable.Value)
 		}
+
+		// If there's a script attatched here, run it now.
+		if wrapper.mib.Script != nil {
+			wrapper.mib.Script.EnrichMetric(idx, oidName, dmr.customBigInt, dmr.customStr, metricsFound)
+		}
 	}
 
 	// Convert to JCFH and pass on.
diff --git a/pkg/inputs/snmp/metrics/interface_metrics.go b/pkg/inputs/snmp/metrics/interface_metrics.go
index 576d1f5..cc18fc9 100644
--- a/pkg/inputs/snmp/metrics/interface_metrics.go
+++ b/pkg/inputs/snmp/metrics/interface_metrics.go
@@ -233,6 +233,11 @@ func (im *InterfaceMetrics) convertToCHF(deltas map[string]map[string]uint64) []
 					format = mib.Format
 				}
 				metrics[k] = kt.MetricInfo{Oid: im.nameOidMap[k], Mib: mib.Mib, Profile: im.profileName, Table: mib.Table, Format: format, PollDur: mib.PollDur}
+
+				// If there's a script attatched here, run it now.
+				if mib.Script != nil {
+					mib.Script.EnrichMetric(strint, k, dst.CustomBigInt, dst.CustomStr, metrics)
+				}
 			} else {
 				metrics[k] = kt.MetricInfo{Oid: im.nameOidMap[k], Profile: im.profileName, Format: kt.CountMetric}
 			}
diff --git a/pkg/inputs/snmp/snmp.go b/pkg/inputs/snmp/snmp.go
index 8452d87..e1a5cda 100644
--- a/pkg/inputs/snmp/snmp.go
+++ b/pkg/inputs/snmp/snmp.go
@@ -13,6 +13,7 @@ import (
 	go_metrics "github.com/kentik/go-metrics"
 	"github.com/kentik/ktranslate"
 	"github.com/kentik/ktranslate/pkg/api"
+	"github.com/kentik/ktranslate/pkg/config"
 	"github.com/kentik/ktranslate/pkg/eggs/logger"
 	"github.com/kentik/ktranslate/pkg/inputs/snmp/metadata"
 	snmp_metrics "github.com/kentik/ktranslate/pkg/inputs/snmp/metrics"
@@ -56,7 +57,7 @@ func init() {
 	flag.BoolVar(&validateMib, "snmp_validate", false, "If true, validate mib profiles and exit.")
 }
 
-func StartSNMPPolls(ctx context.Context, jchfChan chan []*kt.JCHF, metrics *kt.SnmpMetricSet, registry go_metrics.Registry, apic *api.KentikApi, log logger.ContextL, cfg *ktranslate.SNMPInputConfig, resolv *resolv.Resolver) error {
+func StartSNMPPolls(ctx context.Context, jchfChan chan []*kt.JCHF, metrics *kt.SnmpMetricSet, registry go_metrics.Registry, apic *api.KentikApi, log logger.ContextL, cfg *ktranslate.SNMPInputConfig, resolv *resolv.Resolver, confMgr config.ConfigManager) error {
 	snmpFile := cfg.SNMPFile
 	// Do this once here just to see if we need to exit right away.
 	conf, connectTimeout, retries, err := initSnmp(ctx, snmpFile, log)
@@ -96,7 +97,7 @@ func StartSNMPPolls(ctx context.Context, jchfChan chan []*kt.JCHF, metrics *kt.S
 
 	// Now, launch a metadata and metrics server for each configured or discovered device.
 	if conf.Trap == nil || !conf.Trap.TrapOnly { // Unless we are turning off everything but snmp traps.
-		go wrapSnmpPolling(ctx, snmpFile, jchfChan, metrics, registry, apic, log, 0, cfg)
+		go wrapSnmpPolling(ctx, snmpFile, jchfChan, metrics, registry, apic, log, 0, cfg, confMgr)
 	}
 
 	// Run a trap listener?
@@ -146,7 +147,7 @@ func initSnmp(ctx context.Context, snmpFile string, log logger.ContextL) (*kt.Sn
 	return conf, connectTimeout, retries, nil
 }
 
-func wrapSnmpPolling(ctx context.Context, snmpFile string, jchfChan chan []*kt.JCHF, metrics *kt.SnmpMetricSet, registry go_metrics.Registry, apic *api.KentikApi, log logger.ContextL, restartCount int, cfg *ktranslate.SNMPInputConfig) {
+func wrapSnmpPolling(ctx context.Context, snmpFile string, jchfChan chan []*kt.JCHF, metrics *kt.SnmpMetricSet, registry go_metrics.Registry, apic *api.KentikApi, log logger.ContextL, restartCount int, cfg *ktranslate.SNMPInputConfig, confMgr config.ConfigManager) {
 	ctxSnmp, cancel := context.WithCancel(ctx)
 	err := runSnmpPolling(ctxSnmp, snmpFile, jchfChan, metrics, registry, apic, log, restartCount, cfg)
 	if err != nil {
@@ -163,7 +164,7 @@ func wrapSnmpPolling(ctx context.Context, snmpFile string, jchfChan chan []*kt.J
 	c := make(chan os.Signal, 1)
 	signal.Notify(c, kt.SIGUSR2)
 	if v := cfg.DiscoveryIntervalMinutes; v > 0 || runOnStart { // If we are re-running snmp discovery every interval AND/OR running on start, start the ticker here.
-		go RunDiscoOnTimer(ctxSnmp, c, log, v, runOnStart, cfg, apic)
+		go RunDiscoOnTimer(ctxSnmp, c, log, v, runOnStart, cfg, apic, confMgr)
 	}
 
 	// Block here
@@ -172,7 +173,7 @@ func wrapSnmpPolling(ctx context.Context, snmpFile string, jchfChan chan []*kt.J
 	// If we got this signal, redo the snmp system.
 	cancel()
 
-	go wrapSnmpPolling(ctx, snmpFile, jchfChan, metrics, registry, apic, log, restartCount+1, cfg) // Track how many times through here we've been.
+	go wrapSnmpPolling(ctx, snmpFile, jchfChan, metrics, registry, apic, log, restartCount+1, cfg, confMgr) // Track how many times through here we've been.
 }
 
 func runSnmpPolling(ctx context.Context, snmpFile string, jchfChan chan []*kt.JCHF, metrics *kt.SnmpMetricSet, registry go_metrics.Registry, apic *api.KentikApi, log logger.ContextL, restartCount int, cfg *ktranslate.SNMPInputConfig) error {
diff --git a/pkg/kt/snmp.go b/pkg/kt/snmp.go
index b111c1f..08ab852 100644
--- a/pkg/kt/snmp.go
+++ b/pkg/kt/snmp.go
@@ -265,6 +265,7 @@ type SnmpGlobalConfig struct {
 	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.
+	NoDeviceHardcodedOids  bool          `yaml:"no_device_hardcoded_oids"`
 	UserTags      map[string]string      `yaml:"user_tags"`
 	MatchAttr     map[string]string      `yaml:"match_attributes"`
 	ProviderMap   map[string]ProviderMap `yaml:"providers"`
@@ -386,6 +387,7 @@ type Mib struct {
 
 type Enricher interface {
 	EnrichMib(string, string, map[string]interface{}, *LastMetadata)
+	EnrichMetric(string, string, map[string]int64, map[string]string, map[string]MetricInfo)
 }
 
 func (mb *Mib) String() string {
diff --git a/pkg/sinks/ddog/ddog.go b/pkg/sinks/ddog/ddog.go
new file mode 100644
index 0000000..e3d9a3a
--- /dev/null
+++ b/pkg/sinks/ddog/ddog.go
@@ -0,0 +1,131 @@
+package ddog
+
+import (
+	"bytes"
+	"context"
+	"crypto/tls"
+	"flag"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"os"
+
+	go_metrics "github.com/kentik/go-metrics"
+	"github.com/kentik/ktranslate"
+	"github.com/kentik/ktranslate/pkg/eggs/logger"
+	"github.com/kentik/ktranslate/pkg/formats"
+	"github.com/kentik/ktranslate/pkg/kt"
+)
+
+type DDogSink struct {
+	logger.ContextL
+	TargetUrl string
+
+	client   *http.Client
+	tr       *http.Transport
+	registry go_metrics.Registry
+	metrics  *DDogMetric
+	headers  map[string]string
+}
+
+type DDogMetric struct {
+	DeliveryErr go_metrics.Meter
+	DeliveryWin go_metrics.Meter
+}
+
+const (
+	DD_API_KEY = "DD_API_KEY"
+	DD_APP_KEY = "DD_APP_KEY"
+)
+
+func init() {
+	flag.String("ddog_url", "https://api.datadoghq.com/api/v2/series", "URL to post to")
+}
+
+func NewSink(log logger.Underlying, registry go_metrics.Registry, config *ktranslate.DDogSinkConfig) (*DDogSink, error) {
+	ddog := DDogSink{
+		ContextL: logger.NewContextLFromUnderlying(logger.SContext{S: "ddogSink"}, log),
+		registry: registry,
+		metrics: &DDogMetric{
+			DeliveryErr: go_metrics.GetOrRegisterMeter("delivery_errors_ddog", registry),
+			DeliveryWin: go_metrics.GetOrRegisterMeter("delivery_wins_ddog", registry),
+		},
+		headers: map[string]string{
+			"Content-Type": "application/json",
+		},
+		TargetUrl: config.URL,
+	}
+
+	return &ddog, nil
+}
+
+func (s *DDogSink) Init(ctx context.Context, format formats.Format, compression kt.Compression, fmtr formats.Formatter) error {
+	s.tr = &http.Transport{
+		TLSClientConfig: &tls.Config{InsecureSkipVerify: false},
+	}
+	s.client = &http.Client{Transport: s.tr}
+
+	if compression == kt.CompressionGzip {
+		s.headers["Content-Encoding"] = "GZIP"
+	}
+
+	// Fill in some values from envs.
+	apiKey := os.Getenv(DD_API_KEY)
+	if apiKey == "" {
+		return fmt.Errorf("Missing env var %s", DD_API_KEY)
+	}
+	s.headers["DD-API-KEY"] = apiKey
+
+	appKey := os.Getenv(DD_APP_KEY) // @TODO -- does this do anything?
+	if appKey != "" {
+		s.headers["DD-APP-KEY"] = appKey
+	}
+
+	s.Infof("Exporting to DDog at %s", s.TargetUrl)
+	return nil
+}
+
+func (s *DDogSink) Send(ctx context.Context, payload *kt.Output) {
+	go s.sendHttp(ctx, payload.Body)
+}
+
+func (s *DDogSink) Close() {}
+
+func (s *DDogSink) HttpInfo() map[string]float64 {
+	return map[string]float64{
+		"DeliveryErr": s.metrics.DeliveryErr.Rate1(),
+		"DeliveryWin": s.metrics.DeliveryWin.Rate1(),
+	}
+}
+
+func (s *DDogSink) sendHttp(ctx context.Context, payload []byte) {
+	req, err := http.NewRequestWithContext(ctx, "POST", s.TargetUrl, bytes.NewBuffer(payload))
+	if err != nil {
+		s.Errorf("Cannot create HTTP request: %v", err)
+		return
+	}
+
+	for k, v := range s.headers {
+		req.Header.Set(k, v)
+	}
+
+	resp, err := s.client.Do(req)
+	if err != nil {
+		s.Errorf("Cannot write to DDOG: %v, creating new client", err)
+		s.client = &http.Client{Transport: s.tr}
+	} else {
+		defer resp.Body.Close()
+		bdy, err := ioutil.ReadAll(resp.Body)
+		if err != nil {
+			s.Errorf("Cannot get resp body from DDOG: %v", err)
+			s.metrics.DeliveryErr.Mark(1)
+		} else {
+			if resp.StatusCode >= 400 {
+				s.Errorf("Cannot write to DDOG, status code %d, bdy: %s", resp.StatusCode, string(bdy))
+				s.metrics.DeliveryErr.Mark(1)
+			} else {
+				s.metrics.DeliveryWin.Mark(1)
+			}
+		}
+	}
+}
diff --git a/pkg/sinks/sinks.go b/pkg/sinks/sinks.go
index 8e8d9c6..9b51d24 100644
--- a/pkg/sinks/sinks.go
+++ b/pkg/sinks/sinks.go
@@ -10,6 +10,7 @@ import (
 
 	"github.com/kentik/ktranslate/pkg/formats"
 	"github.com/kentik/ktranslate/pkg/kt"
+	"github.com/kentik/ktranslate/pkg/sinks/ddog"
 	"github.com/kentik/ktranslate/pkg/sinks/file"
 	"github.com/kentik/ktranslate/pkg/sinks/gcloud"
 	"github.com/kentik/ktranslate/pkg/sinks/gcppubsub"
@@ -47,6 +48,7 @@ const (
 	GCloudSink         = "gcloud"
 	GCPPubSub          = "gcppubsub"
 	NewRelicMulti      = "new_relic_multi"
+	DDogSink           = "ddog"
 )
 
 func NewSink(sink Sink, log logger.Underlying, registry go_metrics.Registry, tooBig chan int, logTee chan string, config *ktranslate.Config) (SinkImpl, error) {
@@ -59,6 +61,8 @@ func NewSink(sink Sink, log logger.Underlying, registry go_metrics.Registry, too
 		return kafka.NewSink(log, registry, config.KafkaSink)
 	case NewRelicSink:
 		return nr.NewSink(log, registry, tooBig, logTee, config.NewRelicSink)
+	case DDogSink:
+		return ddog.NewSink(log, registry, config.DDogSink)
 	case KentikSink:
 		return kentik.NewSink(log, registry, config)
 	case NetSink:
diff --git a/pkg/util/enrich/builtins.go b/pkg/util/enrich/builtins.go
new file mode 100644
index 0000000..3c53e69
--- /dev/null
+++ b/pkg/util/enrich/builtins.go
@@ -0,0 +1,76 @@
+package enrich
+
+import (
+	"fmt"
+	"regexp"
+	"sync"
+
+	"go.starlark.net/starlark"
+)
+
+// catch(f) evaluates f() and returns its evaluation error message
+// if it failed or None if it succeeded.
+func catch(thread *starlark.Thread, _ *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
+	var fn starlark.Callable
+	if err := starlark.UnpackArgs("catch", args, kwargs, "fn", &fn); err != nil {
+		return nil, err
+	}
+	if _, err := starlark.Call(thread, fn, nil, nil); err != nil {
+		return starlark.String(err.Error()), nil
+	}
+	return starlark.None, nil
+}
+
+// Keep a cache of seen regexps seen to speed things up. Note -- this isn't guarded with a lock, should it be?
+var (
+	reCache = map[string]*regexp.Regexp{}
+	mux     sync.RWMutex
+)
+
+// findAllSubmatch(re, target) will compile re as a regexp and then run findAllSubmatch(target)
+func findAllSubmatch(thread *starlark.Thread, _ *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
+	var re starlark.String
+	var target starlark.String
+	if err := starlark.UnpackArgs("findAllSubmatch", args, kwargs, "re", &re, "target", &target); err != nil {
+		return nil, err
+	}
+
+	// Use cache, assume there's lots of hits and only a few res so don't need a lock here to populate map.
+	reg := re.GoString()
+	mux.RLock()
+	r := reCache[reg]
+	mux.RUnlock()
+	if r == nil {
+		rn, err := regexp.Compile(reg)
+		if err != nil {
+			return starlark.None, err
+		}
+		mux.Lock()
+		reCache[reg] = rn
+		mux.Unlock()
+		r = rn
+	}
+
+	outputs := []starlark.Value{}
+	for _, match := range r.FindAllSubmatch([]byte(target.GoString()), -1) {
+		outputs = append(outputs, starlark.NewList([]starlark.Value{
+			starlark.String(match[0]), starlark.String(match[1]),
+		}))
+	}
+
+	return starlark.NewList(outputs), nil
+}
+
+type builtinMethod func(b *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error)
+
+func builtinAttr(recv starlark.Value, name string, method builtinMethod) (starlark.Value, error) {
+	if method == nil {
+		return starlark.None, fmt.Errorf("no such method '%s'", name)
+	}
+
+	// Allocate a closure over 'method'.
+	impl := func(thread *starlark.Thread, b *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
+		return method(b, args, kwargs)
+	}
+	return starlark.NewBuiltin(name, impl).BindReceiver(recv), nil
+}
diff --git a/pkg/util/enrich/enrich.go b/pkg/util/enrich/enrich.go
index c9e05c4..a26ae8c 100644
--- a/pkg/util/enrich/enrich.go
+++ b/pkg/util/enrich/enrich.go
@@ -73,7 +73,10 @@ func NewEnricher(url string, source string, script string, log logger.Underlying
 				Name:  "kentik enrich",
 				Load:  e.LoadFunc,
 			}
-			builtins := starlark.StringDict{}
+			builtins := starlark.StringDict{
+				"catch":           starlark.NewBuiltin("catch", catch),
+				"findAllSubmatch": starlark.NewBuiltin("findAllSubmatch", findAllSubmatch),
+			}
 
 			program, err := e.sourceProgram(builtins)
 			if err != nil {
@@ -114,6 +117,16 @@ func (e *Enricher) EnrichMib(idx string, key string, attr map[string]interface{}
 	}
 }
 
+func (e *Enricher) EnrichMetric(idx string, key string, ints map[string]int64, strs map[string]string, metrics map[string]kt.MetricInfo) {
+	mm := &MibMetric{ContextL: e}
+	mm.Wrap(idx, key, ints, strs, metrics)
+	_, err := starlark.Call(e.thread, e.globals["main"], starlark.Tuple{mm}, nil)
+	if err != nil {
+		e.Errorf("Cannot run enrich mib script: %v", err)
+		return
+	}
+}
+
 func (e *Enricher) Enrich(ctx context.Context, msgs []*kt.JCHF) ([]*kt.JCHF, error) {
 	if e.doSrc || e.doDst {
 		return e.hashIP(ctx, msgs)
diff --git a/pkg/util/enrich/jchf.go b/pkg/util/enrich/jchf.go
index bc6042d..ce22767 100644
--- a/pkg/util/enrich/jchf.go
+++ b/pkg/util/enrich/jchf.go
@@ -58,7 +58,7 @@ func (m *JCHF) Hash() (uint32, error) {
 
 // AttrNames implements the starlark.HasAttrs interface.
 func (m *JCHF) AttrNames() []string {
-	names := []string{}
+	names := []string{"pop"}
 	for k, _ := range m.flat {
 		names = append(names, k)
 	}
@@ -67,6 +67,10 @@ func (m *JCHF) AttrNames() []string {
 
 // Attr implements the starlark.HasAttrs interface.
 func (m *JCHF) Attr(name string) (starlark.Value, error) {
+	if name == "pop" {
+		return builtinAttr(m, "pop", m.pop)
+	}
+
 	if v, ok := m.flat[name]; ok {
 		switch nv := v.(type) {
 		case kt.Cid:
@@ -131,6 +135,26 @@ func (m *JCHF) Get(key starlark.Value) (v starlark.Value, found bool, err error)
 	return starlark.None, false, errors.New("key must be of type 'str'")
 }
 
+// Delete removes the key and also returns it.
+func (m *JCHF) Delete(key starlark.Value) (v starlark.Value, found bool, err error) {
+	if k, ok := key.(starlark.String); ok {
+		v, err := m.Attr(k.GoString())
+		if err != nil {
+			return starlark.None, false, err
+		}
+
+		// Actually remove the key here.
+		delete(m.metric.CustomBigInt, k.GoString())
+		delete(m.metric.CustomStr, k.GoString())
+		delete(m.metric.CustomInt, k.GoString())
+
+		// And return
+		return v, true, nil
+	}
+
+	return starlark.None, false, errors.New("key must be of type 'str'")
+}
+
 // SetKey implements the starlark.HasSetKey interface to support map update
 // using x[k]=v syntax, like a dictionary.
 func (m *JCHF) SetKey(k, v starlark.Value) error {
@@ -146,6 +170,24 @@ func (m *JCHF) SetKey(k, v starlark.Value) error {
 	return m.SetField(key.GoString(), v)
 }
 
+// Implements the pop method
+func (m *JCHF) pop(b *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
+	var k, d starlark.Value
+	if err := starlark.UnpackPositionalArgs(b.Name(), args, kwargs, 1, &k, &d); err != nil {
+		return starlark.None, fmt.Errorf("%s: %v", b.Name(), err)
+	}
+
+	if v, found, err := m.Delete(k); err != nil {
+		return starlark.None, fmt.Errorf("%s: %v", b.Name(), err)
+	} else if found {
+		return v, nil
+	} else if d != nil {
+		return d, nil
+	}
+
+	return starlark.None, fmt.Errorf("%s: missing key", b.Name())
+}
+
 func setUint64(value starlark.Value) uint64 {
 	switch v := value.(type) {
 	case starlark.Int:
diff --git a/pkg/util/enrich/mib.go b/pkg/util/enrich/mib.go
index 2fffec5..510b39f 100644
--- a/pkg/util/enrich/mib.go
+++ b/pkg/util/enrich/mib.go
@@ -61,7 +61,7 @@ func (m *Mib) Hash() (uint32, error) {
 
 // AttrNames implements the starlark.HasAttrs interface.
 func (m *Mib) AttrNames() []string {
-	names := []string{"this.idx", "this.key"}
+	names := []string{"this.idx", "this.key", "pop"}
 	for k, _ := range m.attr {
 		names = append(names, k)
 	}
@@ -75,6 +75,8 @@ func (m *Mib) Attr(name string) (starlark.Value, error) {
 		return starlark.String(m.idx), nil
 	case "this.key":
 		return starlark.String(m.key), nil
+	case "pop":
+		return builtinAttr(m, "pop", m.pop)
 	default:
 		if v, ok := m.attr[name]; ok {
 			switch nv := v.(type) {
@@ -127,6 +129,8 @@ func (m *Mib) SetField(name string, value starlark.Value) error {
 			if ok {
 				m.attr[name] = ns
 			}
+		case starlark.Float:
+			m.attr[name] = v
 		}
 	}
 
@@ -146,6 +150,24 @@ func (m *Mib) Get(key starlark.Value) (v starlark.Value, found bool, err error)
 	return starlark.None, false, errors.New("key must be of type 'str'")
 }
 
+// Delete removes the key and also returns it.
+func (m *Mib) Delete(key starlark.Value) (v starlark.Value, found bool, err error) {
+	if k, ok := key.(starlark.String); ok {
+		v, err := m.Attr(k.GoString())
+		if err != nil {
+			return starlark.None, false, err
+		}
+
+		// Actually remove the key here.
+		delete(m.attr, k.GoString())
+
+		// And return
+		return v, true, nil
+	}
+
+	return starlark.None, false, errors.New("key must be of type 'str'")
+}
+
 // SetKey implements the starlark.HasSetKey interface to support map update
 // using x[k]=v syntax, like a dictionary.
 func (m *Mib) SetKey(k, v starlark.Value) error {
@@ -160,3 +182,21 @@ func (m *Mib) SetKey(k, v starlark.Value) error {
 
 	return m.SetField(key.GoString(), v)
 }
+
+// Implements the pop method
+func (m *Mib) pop(b *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
+	var k, d starlark.Value
+	if err := starlark.UnpackPositionalArgs(b.Name(), args, kwargs, 1, &k, &d); err != nil {
+		return starlark.None, fmt.Errorf("%s: %v", b.Name(), err)
+	}
+
+	if v, found, err := m.Delete(k); err != nil {
+		return starlark.None, fmt.Errorf("%s: %v", b.Name(), err)
+	} else if found {
+		return v, nil
+	} else if d != nil {
+		return d, nil
+	}
+
+	return starlark.None, fmt.Errorf("%s: missing key", b.Name())
+}
diff --git a/pkg/util/enrich/mib_metric.go b/pkg/util/enrich/mib_metric.go
new file mode 100644
index 0000000..919808a
--- /dev/null
+++ b/pkg/util/enrich/mib_metric.go
@@ -0,0 +1,203 @@
+package enrich
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"go.starlark.net/starlark"
+
+	"github.com/kentik/ktranslate/pkg/eggs/logger"
+	"github.com/kentik/ktranslate/pkg/kt"
+)
+
+type MibMetric struct {
+	logger.ContextL
+	idx     string
+	key     string
+	strs    map[string]string
+	ints    map[string]int64
+	metrics map[string]kt.MetricInfo
+	frozen  bool
+}
+
+// Wrap updates the starlark.Metric to wrap a new telegraf.Metric.
+func (m *MibMetric) Wrap(idx string, key string, ints map[string]int64, strs map[string]string, metrics map[string]kt.MetricInfo) {
+	m.idx = idx
+	m.key = key
+	m.strs = strs
+	m.ints = ints
+	m.metrics = metrics
+	m.frozen = false
+}
+
+// String returns the starlark representation of the Metric.
+//
+// The String function is called by both the repr() and str() functions, and so
+// it behaves more like the repr function would in Python.
+func (m *MibMetric) String() string {
+	buf := new(strings.Builder)
+	buf.WriteString("MibMetric(")
+	buf.WriteString(fmt.Sprintf("Index: %s, ", m.idx))
+	buf.WriteString(fmt.Sprintf("Key: %s, ", m.key))
+	buf.WriteString(fmt.Sprintf("Ints: %v", m.ints))
+	buf.WriteString(fmt.Sprintf("Strings: %v", m.strs))
+	buf.WriteString(")")
+	return buf.String()
+}
+
+func (m *MibMetric) Type() string {
+	return "MibMetric"
+}
+
+func (m *MibMetric) Freeze() {
+	m.frozen = true
+}
+
+func (m *MibMetric) Truth() starlark.Bool {
+	return true
+}
+
+func (m *MibMetric) Hash() (uint32, error) {
+	return 0, errors.New("not hashable")
+}
+
+// AttrNames implements the starlark.HasAttrs interface.
+func (m *MibMetric) AttrNames() []string {
+	names := []string{"this.idx", "this.key", "pop"}
+	for k, _ := range m.strs {
+		names = append(names, k)
+	}
+	for k, _ := range m.ints {
+		names = append(names, k)
+	}
+	return names
+}
+
+// Attr implements the starlark.HasAttrs interface.
+func (m *MibMetric) Attr(name string) (starlark.Value, error) {
+	switch name {
+	case "this.idx":
+		return starlark.String(m.idx), nil
+	case "this.key":
+		return starlark.String(m.key), nil
+	case "pop":
+		return builtinAttr(m, "pop", m.pop)
+	default:
+		if v, ok := m.ints[name]; ok {
+			return starlark.MakeInt64(v), nil
+		}
+		if v, ok := m.strs[name]; ok {
+			return starlark.String(v), nil
+		}
+	}
+
+	// By default, return empty string for all other variables.
+	return starlark.String(""), nil
+}
+
+// SetField implements the starlark.HasSetField interface.
+func (m *MibMetric) SetField(name string, value starlark.Value) error {
+	if m.frozen {
+		return fmt.Errorf("cannot modify frozen metric")
+	}
+
+	switch name {
+	case "this.idx":
+		return starlark.NoSuchAttrError(
+			fmt.Sprintf("cannot assign to field '%s'", name))
+	case "this.key":
+		return starlark.NoSuchAttrError(
+			fmt.Sprintf("cannot assign to field '%s'", name))
+	default:
+		// Copy over the info about this new key from the key which started this process.
+		if m.metrics != nil {
+			if _, ok := m.metrics[name]; !ok {
+				m.metrics[name] = m.metrics[m.key]
+			}
+		}
+
+		switch v := value.(type) {
+		case starlark.String:
+			m.strs[name] = v.GoString()
+		case starlark.Int:
+			ns, ok := v.Int64()
+			if ok {
+				m.ints[name] = ns
+			}
+		case starlark.Float:
+			m.ints[name] = int64(v * 1000.)
+			if met, ok := m.metrics[name]; ok {
+				met.Format = kt.FloatMS
+				m.metrics[name] = met
+			}
+		}
+	}
+
+	return nil
+}
+
+// Get implements the starlark.Mapping interface.
+func (m *MibMetric) Get(key starlark.Value) (v starlark.Value, found bool, err error) {
+	if k, ok := key.(starlark.String); ok {
+		v, err := m.Attr(k.GoString())
+		if err != nil {
+			return starlark.None, false, err
+		}
+		return v, true, nil
+	}
+
+	return starlark.None, false, errors.New("key must be of type 'str'")
+}
+
+// Delete removes the key and also returns it.
+func (m *MibMetric) Delete(key starlark.Value) (v starlark.Value, found bool, err error) {
+	if k, ok := key.(starlark.String); ok {
+		v, err := m.Attr(k.GoString())
+		if err != nil {
+			return starlark.None, false, err
+		}
+
+		// Actually remove the key here.
+		delete(m.ints, k.GoString())
+		delete(m.strs, k.GoString())
+
+		// And return
+		return v, true, nil
+	}
+
+	return starlark.None, false, errors.New("key must be of type 'str'")
+}
+
+// SetKey implements the starlark.HasSetKey interface to support map update
+// using x[k]=v syntax, like a dictionary.
+func (m *MibMetric) SetKey(k, v starlark.Value) error {
+	if m.frozen {
+		return fmt.Errorf("cannot modify frozen metric")
+	}
+
+	key, ok := k.(starlark.String)
+	if !ok {
+		return errors.New("field key must be of type 'str'")
+	}
+
+	return m.SetField(key.GoString(), v)
+}
+
+// Implements the pop method
+func (m *MibMetric) pop(b *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
+	var k, d starlark.Value
+	if err := starlark.UnpackPositionalArgs(b.Name(), args, kwargs, 1, &k, &d); err != nil {
+		return starlark.None, fmt.Errorf("%s: %v", b.Name(), err)
+	}
+
+	if v, found, err := m.Delete(k); err != nil {
+		return starlark.None, fmt.Errorf("%s: %v", b.Name(), err)
+	} else if found {
+		return v, nil
+	} else if d != nil {
+		return d, nil
+	}
+
+	return starlark.None, fmt.Errorf("%s: missing key", b.Name())
+}


Test patch:
diff --git a/pkg/formats/influx/influx_test.go b/pkg/formats/influx/influx_test.go
index f9f59bc..21449c9 100644
--- a/pkg/formats/influx/influx_test.go
+++ b/pkg/formats/influx/influx_test.go
@@ -60,6 +60,31 @@ line2`,
 	assert.Contains(str, "line2")
 }
 
+func TestGetMib(t *testing.T) {
+	assert := assert.New(t)
+
+	namespaceTokenSep = "::"
+	input := map[string]interface{}{"Index": "11", "bgp::foo": "22"}
+	mibName := getMib(input, nil)
+
+	assert.Equal("11", input["index"])
+	assert.Equal("22", input["foo"])
+	assert.Equal(nil, input["bgp::foo"])
+	assert.Equal(nil, input["device_ip"])
+	assert.Equal("device", mibName)
+
+	// Now test out normalizations.
+	mibName = getMib(map[string]interface{}{"mib-table": "mytable", "mib-name": "myname"}, nil)
+	assert.Equal("myname::mytable", mibName)
+
+	mibName = getMib(map[string]interface{}{"mib-table": "mytable::if", "mib-name": "/myname"}, nil)
+	assert.Equal("/myname/mytable::if", mibName)
+
+	mibName = getMib(map[string]interface{}{"mib-table": "mytable::if", "mib-name": "myname/"}, nil)
+	assert.Equal("myname/mytable::if", mibName)
+
+}
+
 const (
 	clean = "the quick red fox jumps over the lazy brown dogs"
 	dirty = `the quick red fox
diff --git a/pkg/inputs/snmp/metadata/device_metadata_test.go b/pkg/inputs/snmp/metadata/device_metadata_test.go
index a6844ad..2e11869 100644
--- a/pkg/inputs/snmp/metadata/device_metadata_test.go
+++ b/pkg/inputs/snmp/metadata/device_metadata_test.go
@@ -26,8 +26,9 @@ func (w testWalker) WalkAll(oid string) ([]gosnmp.SnmpPDU, error) {
 func TestPoll(t *testing.T) {
 	l := lt.NewTestContextL(logger.NilContext, t)
 	conf := &kt.SnmpDeviceConfig{}
+	gconf := &kt.SnmpGlobalConfig{}
 	conf.SetTestWalker(testWalker{err: errors.New("Error"), dm: ""}) // Non nil error
-	dm := NewDeviceMetadata(nil, conf, kt.NewSnmpDeviceMetric(nil, "deviceA"), l)
+	dm := NewDeviceMetadata(nil, gconf, conf, kt.NewSnmpDeviceMetric(nil, "deviceA"), l)
 
 	res, err := dm.poll(context.Background(), nil)
 	assert.NoError(t, err)
@@ -48,16 +49,16 @@ func TestPoll(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, 0, len(res.Tables))
 
-	// Test sysdesc
-	conf.SetTestWalker(testWalker{results: []gosnmp.SnmpPDU{{Value: []byte("sysdesc"), Name: ".1.3.6.1.2.1.1.1.0"}}, dm: ""}) // SysDescr.
+	// Test sysName
+	conf.SetTestWalker(testWalker{results: []gosnmp.SnmpPDU{{Value: []byte("sysName"), Name: ".1.3.6.1.2.1.1.5.0"}}, dm: ""}) // sysName.
 	res, err = dm.poll(context.Background(), nil)
 	assert.NoError(t, err)
-	assert.Equal(t, "sysdesc", res.SysDescr)
+	assert.Equal(t, "sysName", res.SysName)
 
 	meta := map[string]*kt.Mib{
 		"1.1.1": &kt.Mib{Name: "foo"},
 	}
-	dm = NewDeviceMetadata(meta, conf, kt.NewSnmpDeviceMetric(nil, "deviceA"), l)
+	dm = NewDeviceMetadata(meta, gconf, conf, kt.NewSnmpDeviceMetric(nil, "deviceA"), l)
 	conf.SetTestWalker(testWalker{results: []gosnmp.SnmpPDU{{Value: "not a byte-slice", Name: ".1.1.1"}}, dm: ""}) // string value.
 	res, err = dm.poll(context.Background(), nil)
 	assert.NoError(t, err)
@@ -67,4 +68,14 @@ func TestPoll(t *testing.T) {
 	res, err = dm.poll(context.Background(), nil)
 	assert.NoError(t, err)
 	assert.Equal(t, 1, len(res.Tables))
+
+	// Test no hardcoded. Should get nothing here.
+	conf = &kt.SnmpDeviceConfig{}
+	gconf = &kt.SnmpGlobalConfig{NoDeviceHardcodedOids: true}
+	conf.SetTestWalker(testWalker{err: errors.New("Error"), dm: ""}) // Non nil error
+	dm = NewDeviceMetadata(nil, gconf, conf, kt.NewSnmpDeviceMetric(nil, "deviceB"), l)
+	conf.SetTestWalker(testWalker{results: []gosnmp.SnmpPDU{{Value: []byte("sysName"), Name: ".1.3.6.1.2.1.1.5.0"}}, dm: ""}) // sysName.
+	res, err = dm.poll(context.Background(), nil)
+	assert.NoError(t, err)
+	assert.Equal(t, "", res.SysName)
 }
diff --git a/pkg/util/enrich/enrich_test.go b/pkg/util/enrich/enrich_test.go
index aaaea7c..e818e0b 100644
--- a/pkg/util/enrich/enrich_test.go
+++ b/pkg/util/enrich/enrich_test.go
@@ -99,3 +99,53 @@ def main(n):
 		assert.Equal("aaa", evt.CustomStr["foo"])
 	}
 }
+
+func TestCatch(t *testing.T) {
+	testDataPy := string(`
+
+def bad():
+  return 100/0
+
+def main(n):
+  res = catch(bad)
+  return res
+`)
+
+	assert := assert.New(t)
+	l := lt.NewTestContextL(logger.NilContext, t)
+	e, err := NewEnricher("", testDataPy, "", l.GetLogger().GetUnderlyingLogger())
+	assert.Nil(err)
+	assert.NotNil(e)
+
+	out, err := e.Enrich(context.Background(), kt.InputTestingSnmp)
+	assert.Nil(err)
+	assert.NotNil(out)
+}
+
+func TestRE(t *testing.T) {
+	testDataPy := string(`
+def main(n):
+  res = findAllSubmatch("foo(.?)", "seafood fool")
+  n[0]["one"] = res[1][1]
+
+  res = findAllSubmatch("foo(.?)", "seafood fool")
+  n[0]["zero"] = res[1][0]
+
+  res = findAllSubmatch("bar(.?)", "seafood barf")
+  n[0]["bar"] = res[0][0]
+
+  return True
+`)
+
+	assert := assert.New(t)
+	l := lt.NewTestContextL(logger.NilContext, t)
+	e, err := NewEnricher("", testDataPy, "", l.GetLogger().GetUnderlyingLogger())
+	assert.Nil(err)
+	assert.NotNil(e)
+
+	out, err := e.Enrich(context.Background(), kt.InputTestingSnmp)
+	assert.Nil(err)
+	assert.Equal("l", out[0].CustomStr["one"])
+	assert.Equal("fool", out[0].CustomStr["zero"])
+	assert.Equal("barf", out[0].CustomStr["bar"])
+}