Feature/expanded flow sample

sflow/decode.go

@@ -23,6 +23,7 @@ import (
 
 const (
 	dataFlowSample     = 1
+	expandedFlowSample = 3
 	dataCounterSample  = 2
 	standardSflow      = 0
 	rawPacketHeader    = 1
@@ -119,6 +120,12 @@ func decodeFlows(samplesPtr unsafe.Pointer, NumSamples uint32) ([]*FlowSample, e
 				return nil, errors.Wrap(err, "Unable to decode flow sample")
 			}
 			flowSamples = append(flowSamples, fs)
+		} else if sfTypeFormat == expandedFlowSample {
+			fs, err := decodeExpandedFlowSample(samplesPtr)
+			if err != nil {
+				return nil, errors.Wrap(err, "Unable to decode flow sample")
+			}
+			flowSamples = append(flowSamples, fs)
 		}
 
 		samplesPtr = unsafe.Pointer(uintptr(samplesPtr) - uintptr(sampleLength+8))
@@ -131,6 +138,17 @@ func decodeFlowSample(flowSamplePtr unsafe.Pointer) (*FlowSample, error) {
 	flowSamplePtr = unsafe.Pointer(uintptr(flowSamplePtr) - uintptr(sizeOfFlowSampleHeader))
 	fsh := (*FlowSampleHeader)(flowSamplePtr)
 
+	return _decodeFlowSample(flowSamplePtr, fsh)
+}
+
+func decodeExpandedFlowSample(flowSamplePtr unsafe.Pointer) (*FlowSample, error) {
+	flowSamplePtr = unsafe.Pointer(uintptr(flowSamplePtr) - uintptr(sizeOfExpandedFlowSampleHeader))
+	fsh := (*ExpandedFlowSampleHeader)(flowSamplePtr).toFlowSampleHeader()
+
+	return _decodeFlowSample(flowSamplePtr, fsh)
+}
+
+func _decodeFlowSample(flowSamplePtr unsafe.Pointer, fsh *FlowSampleHeader) (*FlowSample, error) {
 	var rph *RawPacketHeader
 	var rphd unsafe.Pointer
 	var erd *ExtendedRouterData
@@ -155,7 +173,7 @@ func decodeFlowSample(flowSamplePtr unsafe.Pointer) (*FlowSample, error) {
 			case extendedSwitchData:
 
 			default:
-				log.Infof("Unknown sfTypeFormat\n")
+				log.Infof("Unknown sfTypeFormat %d\n", sfTypeFormat)
 			}
 
 		}

sflow/packet.go

@@ -35,6 +35,7 @@ var (
 	sizeOfHeaderTop                = unsafe.Sizeof(headerTop{})
 	sizeOfHeaderBottom             = unsafe.Sizeof(headerBottom{})
 	sizeOfFlowSampleHeader         = unsafe.Sizeof(FlowSampleHeader{})
+	sizeOfExpandedFlowSampleHeader = unsafe.Sizeof(ExpandedFlowSampleHeader{})
 	sizeOfRawPacketHeader          = unsafe.Sizeof(RawPacketHeader{})
 	sizeofExtendedRouterData       = unsafe.Sizeof(ExtendedRouterData{})
 	sizeOfextendedRouterDataTop    = unsafe.Sizeof(extendedRouterDataTop{})
@@ -66,11 +67,12 @@ type headerBottom struct {
 
 // FlowSample is an sflow version 5 flow sample
 type FlowSample struct {
-	FlowSampleHeader   *FlowSampleHeader
-	RawPacketHeader    *RawPacketHeader
-	Data               unsafe.Pointer
-	DataLen            uint32
-	ExtendedRouterData *ExtendedRouterData
+	FlowSampleHeader         *FlowSampleHeader
+	ExpandedFlowSampleHeader *ExpandedFlowSampleHeader
+	RawPacketHeader          *RawPacketHeader
+	Data                     unsafe.Pointer
+	DataLen                  uint32
+	ExtendedRouterData       *ExtendedRouterData
 }
 
 // FlowSampleHeader is an sflow version 5 flow sample header
@@ -87,6 +89,38 @@ type FlowSampleHeader struct {
 	EnterpriseType     uint32
 }
 
+// ExpandedFlowSampleHeader is an sflow version 5 flow expanded sample header
+type ExpandedFlowSampleHeader struct {
+	FlowRecord         uint32
+	OutputIf           uint32
+	_                  uint32
+	InputIf            uint32
+	_                  uint32
+	DroppedPackets     uint32
+	SamplePool         uint32
+	SamplingRate       uint32
+	SourceIDClassIndex uint32
+	_                  uint32
+	SequenceNumber     uint32
+	SampleLength       uint32
+	EnterpriseType     uint32
+}
+
+func (e *ExpandedFlowSampleHeader) toFlowSampleHeader() *FlowSampleHeader {
+	return &FlowSampleHeader{
+		FlowRecord:         e.FlowRecord,
+		OutputIf:           e.OutputIf,
+		InputIf:            e.InputIf,
+		DroppedPackets:     e.DroppedPackets,
+		SamplePool:         e.SamplePool,
+		SamplingRate:       e.SamplingRate,
+		SourceIDClassIndex: e.SourceIDClassIndex,
+		SequenceNumber:     e.SequenceNumber,
+		SampleLength:       e.SampleLength,
+		EnterpriseType:     e.EnterpriseType,
+	}
+}
+
 // RawPacketHeader is a raw packet header
 type RawPacketHeader struct {
 	OriginalPacketLength uint32

sfserver/sfserver.go

@@ -161,10 +161,9 @@ func (sfs *SflowServer) processPacket(agent net.IP, buffer []byte) {
 		// We're updating the sampleCache to allow the forntend to show current sampling rates
 		sfs.sampleRateCache.Set(agent, uint64(fs.FlowSampleHeader.SamplingRate))
 
-		if fs.ExtendedRouterData == nil {
-			continue
+		if fs.ExtendedRouterData != nil {
+			fl.NextHop = fs.ExtendedRouterData.NextHop
 		}
-		fl.NextHop = fs.ExtendedRouterData.NextHop
 
 		sfs.processEthernet(ether.EtherType, fs, fl)