Finished sampling rate option support in Netflow v9 and moved sampler…

…ate configuration to config file

config.yml.example

@@ -30,4 +30,5 @@ bgp_augmentation:
 agents:
   - name: "bb01.fra01"
     ip_address: "127.0.0.1"
-    snmp_community: "public"
+    snmp_community: "public"
+    samplerate: 1000

config/config.go

@@ -45,12 +45,14 @@ type Agent struct {
 	Name          *string `yaml:"name"`
 	IPAddress     *string `yaml:"ip_address"`
 	SNMPCommunity *string `yaml:"snmp_community"`
+	SampleRate    *uint64 `yaml:"sample_rate"`
 }
 
 var (
 	dfltAggregationPeriod    = int64(60)
 	dfltDefaultSNMPCommunity = "public"
 	dfltDebug                = 0
+	dfltSampleRate           = uint64(1)
 	dfltCompressionLevel     = 6
 	dfltDataDir              = "data"
 	dfltAnonymize            = false
@@ -195,6 +197,9 @@ func (cfg *Config) defaults() {
 			if agent.SNMPCommunity == nil {
 				cfg.Agents[key].SNMPCommunity = strPtr(*cfg.DefaultSNMPCommunity)
 			}
+			if agent.SampleRate == nil {
+				cfg.Agents[key].SampleRate = uint64Ptr(dfltSampleRate)
+			}
 		}
 	}
 }
@@ -203,6 +208,10 @@ func bgpPtr(bgp BGPAugment) *BGPAugment {
 	return &bgp
 }
 
+func uint64Ptr(x uint64) *uint64 {
+	return &x
+}
+
 func srvPtr(srv Server) *Server {
 	return &srv
 }

intfmapper/intfmapper.go

@@ -61,7 +61,7 @@ func (m *Mapper) startRenewWorkers() {
 				time.Sleep(time.Second * time.Duration(m.renewInterval))
 				err := m.renewMapping(agent)
 				if err != nil {
-					glog.Infof("Unable to renew interface mapping for %s: %v", agent.Name, err)
+					glog.Infof("Unable to renew interface mapping for %s: %v", *agent.Name, err)
 				}
 			}
 		}(agent)

nf9/decode.go

@@ -101,28 +101,31 @@ func decodeOption(packet *Packet, end unsafe.Pointer, size uintptr, remote net.I
 	for uintptr(end) > min {
 		headerPtr := unsafe.Pointer(uintptr(end) - sizeOfOptionsTemplateRecordHeader)
 
-		tmplRecs := &OptionsTemplateRecords{}
-		tmplRecs.Header = (*OptionsTemplateRecordHeader)(unsafe.Pointer(headerPtr))
+		tmplRecs := &TemplateRecords{}
+		hdr := (*OptionsTemplateRecordHeader)(unsafe.Pointer(headerPtr))
+		tmplRecs.Header = &TemplateRecordHeader{TemplateID: hdr.TemplateID}
 		tmplRecs.Packet = packet
 		tmplRecs.Records = make([]*TemplateRecord, 0, numPreAllocRecs)
 
 		ptr := headerPtr
 		// Process option scopes
-		for i := uint16(0); i < tmplRecs.Header.OptionScopeLength/uint16(sizeOfOptionScope); i++ {
+		for i := uint16(0); i < hdr.OptionScopeLength/uint16(sizeOfOptionScope); i++ {
 			optScope := (*OptionScope)(ptr)
 			tmplRecs.OptionScopes = append(tmplRecs.OptionScopes, optScope)
 			ptr = unsafe.Pointer(uintptr(ptr) - sizeOfOptionScope)
 		}
 
 		// Process option fields
-		for i := uint16(0); i < tmplRecs.Header.OptionLength/uint16(sizeOfTemplateRecord); i++ {
+		for i := uint16(0); i < hdr.OptionLength/uint16(sizeOfTemplateRecord); i++ {
 			opt := (*TemplateRecord)(ptr)
 			tmplRecs.Records = append(tmplRecs.Records, opt)
 			ptr = unsafe.Pointer(uintptr(ptr) - sizeOfTemplateRecord)
 		}
 
-		packet.OptionsTemplates = append(packet.OptionsTemplates, tmplRecs)
-		end = unsafe.Pointer(uintptr(end) - uintptr(tmplRecs.Header.OptionScopeLength) - uintptr(tmplRecs.Header.OptionLength) - sizeOfOptionsTemplateRecordHeader)
+		//packet.OptionsTemplates = append(packet.OptionsTemplates, tmplRecs)
+		packet.Templates = append(packet.Templates, tmplRecs)
+
+		end = unsafe.Pointer(uintptr(end) - uintptr(hdr.OptionScopeLength) - uintptr(hdr.OptionLength) - sizeOfOptionsTemplateRecordHeader)
 	}
 }
 

nf9/packet.go

@@ -199,10 +199,6 @@ type Packet struct {
 	// representing a template found in this packet.
 	Templates []*TemplateRecords
 
-	// A slice of pointers to OptionsTemplates. Each element is instance of OptionsTemplates
-	// representing a option found in this packet.
-	OptionsTemplates []*OptionsTemplateRecords
-
 	// Buffer is a slice pointing to the original byte array that this packet was decoded from.
 	// This field is only populated if debug level is at least 2
 	Buffer []byte
@@ -216,12 +212,6 @@ func (p *Packet) GetTemplateRecords() []*TemplateRecords {
 	return p.Templates
 }
 
-// GetOptionsTemplateRecords returns a list of all Option Template Records in the packet.
-// Option Template Records can be used to decode Option Data FlowSets to Data Records.
-func (p *Packet) GetOptionsTemplateRecords() []*OptionsTemplateRecords {
-	return p.OptionsTemplates
-}
-
 // DataFlowSets generate a list of all Data FlowSets in the packet. If matched
 // with appropriate templates Data FlowSets can be decoded to Data Records or
 // Options Data Records.

nf9/templates.go

@@ -22,7 +22,6 @@ var (
 	sizeOfTemplateRecordHeader        = unsafe.Sizeof(TemplateRecordHeader{})
 	sizeOfOptionsTemplateRecordHeader = unsafe.Sizeof(OptionsTemplateRecordHeader{})
 	sizeOfOptionScope                 = unsafe.Sizeof(OptionScope{})
-	sizeOfOptionsTemplateRecords      = unsafe.Sizeof(OptionsTemplateRecords{})
 )
 
 // TemplateRecordHeader represents the header of a template record
@@ -64,22 +63,10 @@ type OptionsTemplateRecordHeader struct {
 type TemplateRecords struct {
 	Header *TemplateRecordHeader
 
-	// List of fields in this Template Record.
-	Records []*TemplateRecord
-
-	Packet *Packet
-
-	Values [][]byte
-}
-
-// OptionsTemplateRecords is a single options template that describtes an options Flow Record
-type OptionsTemplateRecords struct {
-	Header *OptionsTemplateRecordHeader
-
 	// List of scopes
 	OptionScopes []*OptionScope
 
-	// List of fields
+	// List of fields in this Template Record.
 	Records []*TemplateRecord
 
 	Packet *Packet
@@ -121,7 +108,7 @@ var sizeOfTemplateRecord = unsafe.Sizeof(TemplateRecord{})
 
 // DecodeFlowSet uses current TemplateRecord to decode data in Data FlowSet to
 // a list of Flow Data Records.
-func (dtpl *TemplateRecords) DecodeFlowSet(set FlowSet) (list []FlowDataRecord) {
+/*func (dtpl *TemplateRecords) DecodeFlowSet(set FlowSet) (list []FlowDataRecord) {
 	if set.Header.FlowSetID != dtpl.Header.TemplateID {
 		return nil
 	}
@@ -145,6 +132,32 @@ func (dtpl *TemplateRecords) DecodeFlowSet(set FlowSet) (list []FlowDataRecord)
 		n = n - count
 	}
 
+	return
+}*/
+
+// DecodeFlowSet uses current TemplateRecord to decode data in Data FlowSet to
+// a list of Flow Data Records.
+func DecodeFlowSet(templateRecords []*TemplateRecord, set FlowSet) (list []FlowDataRecord) {
+	var record FlowDataRecord
+
+	// Pre-allocate some room for flows
+	list = make([]FlowDataRecord, 0, numPreAllocFlowDataRecs)
+
+	// Assume total record length must be >= 4, otherwise it is impossible
+	// to distinguish between padding and new record. Padding MUST be
+	// supported.
+	n := len(set.Flows)
+	count := 0
+
+	for n >= 4 {
+		record.Values, count = parseFieldValues(set.Flows[0:n], templateRecords)
+		if record.Values == nil {
+			return
+		}
+		list = append(list, record)
+		n = n - count
+	}
+
 	return
 }