Repository: kentik/ktranslate
Pull request #182 (2021-11-08 19:27:52)

Allowing a (very raw) version of custom ipfix templates to be used
Enables things like

```
netflowv9:
  mapping:
    - field: 7
      destination: CustomInteger1
    - field: 11
      destination: CustomInteger2
sflow:
  mapping:
    - layer: 4 # Layer 4: TCP or UDP
      offset: 0 # Source port
      length: 16 # 2 bytes
      destination: CustomInteger1
    - layer: 4
      offset: 16 # Destination port
      length: 16 # 2 bytes
      destination: CustomInteger2
```
To pass along arbitrary values. 

Also removing non common flow fields from the default field set to speed up processing. 




Patch:
diff --git a/pkg/inputs/flow/flow.go b/pkg/inputs/flow/flow.go
index 839da66..04ca7e8 100644
--- a/pkg/inputs/flow/flow.go
+++ b/pkg/inputs/flow/flow.go
@@ -5,6 +5,7 @@ import (
 	"flag"
 	"fmt"
 	"net/http"
+	"os"
 
 	go_metrics "github.com/kentik/go-metrics"
 
@@ -24,7 +25,9 @@ const (
 	Netflow5            = "netflow5"
 	Netflow9            = "netflow9"
 
-	defaultFields = "Type,TimeReceived,SequenceNum,SamplingRate,SamplerAddress,TimeFlowStart,TimeFlowEnd,Bytes,Packets,SrcAddr,DstAddr,Etype,Proto,SrcPort,DstPort,InIf,OutIf,SrcMac,DstMac,SrcVlan,DstVlan,VlanId,IngressVrfID,EgressVrfID,IPTos,ForwardingStatus,IPTTL,TCPFlags,IcmpType,IcmpCode,IPv6FlowLabel,FragmentId,FragmentOffset,BiFlowDirection,SrcAS,DstAS,NextHop,NextHopAS,SrcNet,DstNet,HasMPLS,MPLSCount,MPLS1TTL,MPLS1Label,MPLS2TTL,MPLS2Label,MPLS3TTL,MPLS3Label,MPLSLastTTL,MPLSLastLabel"
+	// These can be any of
+	fullFieldList = "Type,TimeReceived,SequenceNum,SamplingRate,SamplerAddress,TimeFlowStart,TimeFlowEnd,Bytes,Packets,SrcAddr,DstAddr,Etype,Proto,SrcPort,DstPort,InIf,OutIf,SrcMac,DstMac,SrcVlan,DstVlan,VlanId,IngressVrfID,EgressVrfID,IPTos,ForwardingStatus,IPTTL,TCPFlags,IcmpType,IcmpCode,IPv6FlowLabel,FragmentId,FragmentOffset,BiFlowDirection,SrcAS,DstAS,NextHop,NextHopAS,SrcNet,DstNet,HasMPLS,MPLSCount,MPLS1TTL,MPLS1Label,MPLS2TTL,MPLS2Label,MPLS3TTL,MPLS3Label,MPLSLastTTL,MPLSLastLabel,CustomInteger1,CustomInteger2,CustomBytes1,CustomBytes2"
+	defaultFields = "TimeReceived,SamplingRate,Bytes,Packets,SrcAddr,DstAddr,Proto,SrcPort,DstPort,InIf,OutIf,SrcVlan,DstVlan,TCPFlags,SrcAS,DstAS"
 )
 
 var (
@@ -32,8 +35,9 @@ var (
 	Port          = flag.Int("nf.port", 9995, "Sflow/NetFlow/IPFIX listening port")
 	Reuse         = flag.Bool("nf.reuserport", false, "Enable so_reuseport for Sflow/NetFlow/IPFIX")
 	Workers       = flag.Int("nf.workers", 1, "Number of workers per flow collector")
-	MessageFields = flag.String("nf.message.fields", defaultFields, "The list of fields to include in flow messages")
+	MessageFields = flag.String("nf.message.fields", defaultFields, "The list of fields to include in flow messages. Can be any of "+fullFieldList)
 	PromPath      = flag.String("nf.prom.listen", "", "Run a promethues metrics collector here")
+	MappingFile   = flag.String("nf.mapping", "", "Configuration file for custom netflow mappings")
 )
 
 func NewFlowSource(ctx context.Context, proto FlowSource, maxBatchSize int, log logger.Underlying, registry go_metrics.Registry, jchfChan chan []*kt.JCHF, apic *api.KentikApi) (*KentikDriver, error) {
@@ -48,11 +52,28 @@ func NewFlowSource(ctx context.Context, proto FlowSource, maxBatchSize int, log
 		}
 	}()
 
+	// Allow processing of custom ipfix templates here.
+	var config utils.ProducerConfig
+	if *MappingFile != "" {
+		f, err := os.Open(*MappingFile)
+		if err != nil {
+			kt.Errorf("Cannot load netflow mapping file: %v", err)
+			return nil, err
+		}
+		config, err = utils.LoadMapping(f)
+		f.Close()
+		if err != nil {
+			kt.Errorf("Invalid yaml for netflow mapping file: %v", err)
+			return nil, err
+		}
+	}
+
 	switch proto {
 	case Ipfix, Netflow9:
 		sNF := &utils.StateNetFlow{
 			Format: kt,
 			Logger: &KentikLog{l: kt},
+			Config: config,
 		}
 		go func() { // Let this run, returning flow into the kentik transport struct
 			err := sNF.FlowRoutine(*Workers, *Addr, *Port, *Reuse)
@@ -65,6 +86,7 @@ func NewFlowSource(ctx context.Context, proto FlowSource, maxBatchSize int, log
 		sSF := &utils.StateSFlow{
 			Format: kt,
 			Logger: &KentikLog{l: kt},
+			Config: config,
 		}
 		go func() { // Let this run, returning flow into the kentik transport struct
 			err := sSF.FlowRoutine(*Workers, *Addr, *Port, *Reuse)
diff --git a/pkg/inputs/flow/format.go b/pkg/inputs/flow/format.go
index 3c6c381..898eead 100644
--- a/pkg/inputs/flow/format.go
+++ b/pkg/inputs/flow/format.go
@@ -236,6 +236,14 @@ func (t *KentikDriver) toJCHF(fmsg *flowmessage.FlowMessage) *kt.JCHF {
 			in.CustomBigInt[field] = int64(fmsg.MPLSLastTTL)
 		case "MPLSLastLabel":
 			in.CustomBigInt[field] = int64(fmsg.MPLSLastLabel)
+		case "CustomInteger1":
+			in.CustomBigInt[field] = int64(fmsg.CustomInteger1)
+		case "CustomInteger2":
+			in.CustomBigInt[field] = int64(fmsg.CustomInteger2)
+		case "CustomBytes1":
+			in.CustomStr[field] = fmt.Sprintf("%.2x", fmsg.CustomBytes1)
+		case "CustomBytes2":
+			in.CustomStr[field] = fmt.Sprintf("%.2x", fmsg.CustomBytes2)
 		}
 	}
 
diff --git a/pkg/inputs/snmp/metadata/poll.go b/pkg/inputs/snmp/metadata/poll.go
index 1c38225..8fa9e3a 100644
--- a/pkg/inputs/snmp/metadata/poll.go
+++ b/pkg/inputs/snmp/metadata/poll.go
@@ -11,6 +11,7 @@ import (
 
 	"github.com/kentik/ktranslate/pkg/eggs/logger"
 	"github.com/kentik/ktranslate/pkg/inputs/snmp/mibs"
+	snmp_util "github.com/kentik/ktranslate/pkg/inputs/snmp/util"
 	"github.com/kentik/ktranslate/pkg/kt"
 	"github.com/kentik/ktranslate/pkg/util/tick"
 )
@@ -247,8 +248,12 @@ func (p *Poller) toFlows(dd *kt.DeviceData) ([]*kt.JCHF, error) {
 
 	// Also any device level tags
 	for k, v := range p.conf.UserTags {
-		dst.CustomStr[k] = v
-		dst.CustomMetrics[k] = kt.MetricInfo{Table: kt.DeviceTagTable}
+		key := k
+		if !strings.HasPrefix(key, snmp_util.NRUserTagPrefix) {
+			key = snmp_util.NRUserTagPrefix + k
+		}
+		dst.CustomStr[key] = v
+		dst.CustomMetrics[key] = kt.MetricInfo{Table: kt.DeviceTagTable}
 	}
 
 	for intr, id := range dd.InterfaceData {
diff --git a/pkg/inputs/snmp/metrics/device_metrics.go b/pkg/inputs/snmp/metrics/device_metrics.go
index 3277bf6..f6e56a2 100644
--- a/pkg/inputs/snmp/metrics/device_metrics.go
+++ b/pkg/inputs/snmp/metrics/device_metrics.go
@@ -247,6 +247,11 @@ func (dm *DeviceMetrics) pollFromConfig(ctx context.Context, server *gosnmp.GoSN
 					dst.CustomBigInt["MemoryUtilization"] = int64(float32(memoryUsed) / float32(memoryTotal) * 100)
 					dst.CustomMetrics["MemoryUtilization"] = kt.MetricInfo{Oid: "computed", Mib: "computed", Profile: dm.profileName}
 				}
+			} 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}
+				}
 			}
 		}
 		flows = append(flows, dst)
@@ -309,7 +314,11 @@ func (dm *DeviceMetrics) GetStatusFlows() []*kt.JCHF {
 	dst.Timestamp = time.Now().Unix()
 	dst.CustomMetrics = map[string]kt.MetricInfo{"PollingHealth": kt.MetricInfo{Oid: "computed", Mib: "computed", Profile: dm.profileName}}
 	for k, v := range dm.conf.UserTags {
-		dst.CustomStr[k] = v
+		key := k
+		if !strings.HasPrefix(key, snmp_util.NRUserTagPrefix) {
+			key = snmp_util.NRUserTagPrefix + k
+		}
+		dst.CustomStr[key] = v
 	}
 	if dst.Provider == kt.ProviderDefault { // Add this to trigger a UI element.
 		dst.CustomStr["profile_message"] = kt.DefaultProfileMessage
diff --git a/pkg/inputs/snmp/mibs/profile.go b/pkg/inputs/snmp/mibs/profile.go
index 83402c5..ad23637 100644
--- a/pkg/inputs/snmp/mibs/profile.go
+++ b/pkg/inputs/snmp/mibs/profile.go
@@ -33,13 +33,14 @@ type Tag struct {
 }
 
 type MIB struct {
-	Mib        string `yaml:"MIB,omitempty"`
-	Table      OID    `yaml:"table,omitempty"`
-	Symbols    []OID  `yaml:"symbols,omitempty"`
-	MetricTags []Tag  `yaml:"metric_tags,omitempty"`
-	ForcedType string `yaml:"forced_type,omitempty"`
-	Symbol     OID    `yaml:"symbol,omitempty"`
-	sortKey    string
+	Mib          string `yaml:"MIB,omitempty"`
+	Table        OID    `yaml:"table,omitempty"`
+	Symbols      []OID  `yaml:"symbols,omitempty"`
+	MetricTags   []Tag  `yaml:"metric_tags,omitempty"`
+	ForcedType   string `yaml:"forced_type,omitempty"`
+	Symbol       OID    `yaml:"symbol,omitempty"`
+	sortKey      string
+	fromExtended bool
 }
 
 type Device struct {
@@ -102,7 +103,10 @@ func (p *Profile) extend(extends map[string]*Profile) error {
 }
 
 func (p *Profile) merge(ep *Profile) {
-	p.Metrics = append(p.Metrics, ep.Metrics...)
+	for _, m := range ep.Metrics {
+		m.fromExtended = true
+		p.Metrics = append(p.Metrics, m)
+	}
 	p.MetricTags = append(p.MetricTags, ep.MetricTags...)
 }
 
@@ -350,14 +354,15 @@ func (p *Profile) GetMetrics(enabledMibs []string, counterTimeSec int) (map[stri
 		// TODO -- so we want to collase Symbol and Symbols?
 		if metric.Symbol.Oid != "" {
 			mib := &kt.Mib{
-				Oid:        metric.Symbol.Oid,
-				Name:       metric.Symbol.Name,
-				Type:       otype,
-				Enum:       metric.Symbol.Enum,
-				Tag:        metric.Symbol.Tag,
-				Conversion: metric.Symbol.Conversion,
-				Mib:        metric.Mib,
-				Table:      metric.Table.GetTableName(),
+				Oid:          metric.Symbol.Oid,
+				Name:         metric.Symbol.Name,
+				Type:         otype,
+				Enum:         metric.Symbol.Enum,
+				Tag:          metric.Symbol.Tag,
+				Conversion:   metric.Symbol.Conversion,
+				Mib:          metric.Mib,
+				Table:        metric.Table.GetTableName(),
+				FromExtended: metric.fromExtended,
 			}
 			if len(mib.Enum) > 0 {
 				mib.EnumRev = make(map[int64]string)
@@ -387,14 +392,15 @@ func (p *Profile) GetMetrics(enabledMibs []string, counterTimeSec int) (map[stri
 
 		for _, s := range metric.Symbols {
 			mib := &kt.Mib{
-				Oid:        s.Oid,
-				Name:       s.Name,
-				Type:       otype,
-				Enum:       s.Enum,
-				Tag:        s.Tag,
-				Conversion: s.Conversion,
-				Mib:        metric.Mib,
-				Table:      metric.Table.GetTableName(),
+				Oid:          s.Oid,
+				Name:         s.Name,
+				Type:         otype,
+				Enum:         s.Enum,
+				Tag:          s.Tag,
+				Conversion:   s.Conversion,
+				Mib:          metric.Mib,
+				Table:        metric.Table.GetTableName(),
+				FromExtended: metric.fromExtended,
 			}
 			if len(mib.Enum) > 0 {
 				mib.EnumRev = make(map[int64]string)
@@ -670,18 +676,24 @@ 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) {
 	seenNames := map[string]string{}
+	fromExtended := map[string]bool{}
 
 	for oid, mib := range mibs {
-		oidName := mib.Name
-		if mib.Tag != "" {
-			oidName = mib.Tag
-		}
+		oidName := mib.GetName()
 		if _, ok := seenNames[oidName]; !ok { // We haven't seen this name yet so mark that we have it.
 			seenNames[oidName] = oid
+			fromExtended[oidName] = mib.FromExtended
 		} else {
-			// There's a conflict. Keep the longest oid.
-			if len(oid) > len(seenNames[oidName]) {
+			// There's a conflict. If this mib is not fromextended and the other one is, swap.
+			if fromExtended[oidName] && !mib.FromExtended {
 				seenNames[oidName] = oid
+				fromExtended[oidName] = mib.FromExtended
+			} else if mib.FromExtended && !fromExtended[oidName] { // Else, if this mib is fromextended, keep the other one.
+				// Noop.
+			} else {
+				if len(oid) > len(seenNames[oidName]) { // Finally, keep the mib with the longest oid.
+					seenNames[oidName] = oid
+				}
 			}
 		}
 	}
diff --git a/pkg/inputs/snmp/util/util.go b/pkg/inputs/snmp/util/util.go
index c3f519f..3f566df 100644
--- a/pkg/inputs/snmp/util/util.go
+++ b/pkg/inputs/snmp/util/util.go
@@ -21,6 +21,8 @@ const (
 	CONV_HWADDR   = "hwaddr"
 	CONV_POWERSET = "powerset_status"
 	CONV_HEXTOINT = "hextoint"
+
+	NRUserTagPrefix = "tags."
 )
 
 var (
diff --git a/pkg/kt/snmp.go b/pkg/kt/snmp.go
index f5ba46e..d929101 100644
--- a/pkg/kt/snmp.go
+++ b/pkg/kt/snmp.go
@@ -260,19 +260,20 @@ const (
 )
 
 type Mib struct {
-	Oid        string
-	Name       string
-	Type       Oidtype
-	Extra      string
-	Tag        string
-	Enum       map[string]int64
-	EnumRev    map[int64]string
-	Conversion string
-	Mib        string
-	Table      string
-	PollDur    time.Duration
-	MatchAttr  map[string]*regexp.Regexp
-	lastPoll   time.Time
+	Oid          string
+	Name         string
+	Type         Oidtype
+	Extra        string
+	Tag          string
+	Enum         map[string]int64
+	EnumRev      map[int64]string
+	Conversion   string
+	Mib          string
+	Table        string
+	PollDur      time.Duration
+	MatchAttr    map[string]*regexp.Regexp
+	lastPoll     time.Time
+	FromExtended bool
 }
 
 func (mb *Mib) String() string {


Test patch:
diff --git a/pkg/inputs/snmp/mibs/profile_test.go b/pkg/inputs/snmp/mibs/profile_test.go
index 5e04559..afe6fbc 100644
--- a/pkg/inputs/snmp/mibs/profile_test.go
+++ b/pkg/inputs/snmp/mibs/profile_test.go
@@ -124,12 +124,25 @@ 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"},
-		"1.3.6.1.4.1.9.9.305.1.1.1":       &kt.Mib{Tag: "CPU"},
+		".1.3.6.1.4.1.9.9.305.1.1.1":      &kt.Mib{Tag: "CPU"},
 		".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)
 	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))
+
+	mibs = map[string]*kt.Mib{
+		".1.3.6.1.4.1.9.9.48.1.1.1.5":     &kt.Mib{Tag: "MemoryUsed"},
+		".1.3.6.1.4.1.9.9.305.1.1.1":      &kt.Mib{Tag: "CPU"},
+		".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)
+	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))
 }
 
 func TestProfileName(t *testing.T) {