Merge pull request #81 from aams-telefonica/feature/timeflow-start-end-ms

Add TimeFlowStartMs and TimeFlowEndMs to Netflow v5, v9, v10 (IPFIX)

format/common/text.go

@@ -58,6 +58,8 @@ var (
 		"SamplerAddress",
 		"TimeFlowStart",
 		"TimeFlowEnd",
+		"TimeFlowStartMs",
+		"TimeFlowEndMs",
 		"Bytes",
 		"Packets",
 		"SrcAddr",
@@ -102,6 +104,8 @@ var (
 		FORMAT_TYPE_INTEGER,
 		FORMAT_TYPE_INTEGER,
 		FORMAT_TYPE_INTEGER,
+		FORMAT_TYPE_INTEGER,
+		FORMAT_TYPE_INTEGER,
 		FORMAT_TYPE_IP,
 		FORMAT_TYPE_IP,
 		FORMAT_TYPE_INTEGER,

pb/flow.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.26.0
+// 	protoc-gen-go v1.28.0
-// 	protoc        v3.17.2
+// 	protoc        v3.12.1
 // source: pb/flow.proto
 
 package flowpb
@@ -90,6 +90,8 @@ type FlowMessage struct {
 	// Found inside packet
 	TimeFlowStart   uint64 `protobuf:"varint,38,opt,name=TimeFlowStart,proto3" json:"TimeFlowStart,omitempty"`
 	TimeFlowEnd     uint64 `protobuf:"varint,5,opt,name=TimeFlowEnd,proto3" json:"TimeFlowEnd,omitempty"`
+	TimeFlowStartMs uint64 `protobuf:"varint,63,opt,name=TimeFlowStartMs,proto3" json:"TimeFlowStartMs,omitempty"`
+	TimeFlowEndMs   uint64 `protobuf:"varint,64,opt,name=TimeFlowEndMs,proto3" json:"TimeFlowEndMs,omitempty"`
 	// Size of the sampled packet
 	Bytes   uint64 `protobuf:"varint,9,opt,name=Bytes,proto3" json:"Bytes,omitempty"`
 	Packets uint64 `protobuf:"varint,10,opt,name=Packets,proto3" json:"Packets,omitempty"`
@@ -151,8 +153,14 @@ type FlowMessage struct {
 	// Custom allocations
 	CustomInteger1 uint64 `protobuf:"varint,1001,opt,name=CustomInteger1,proto3" json:"CustomInteger1,omitempty"`
 	CustomInteger2 uint64 `protobuf:"varint,1002,opt,name=CustomInteger2,proto3" json:"CustomInteger2,omitempty"`
+	CustomInteger3 uint64 `protobuf:"varint,1003,opt,name=CustomInteger3,proto3" json:"CustomInteger3,omitempty"`
+	CustomInteger4 uint64 `protobuf:"varint,1004,opt,name=CustomInteger4,proto3" json:"CustomInteger4,omitempty"`
+	CustomInteger5 uint64 `protobuf:"varint,1005,opt,name=CustomInteger5,proto3" json:"CustomInteger5,omitempty"`
 	CustomBytes1   []byte `protobuf:"bytes,1011,opt,name=CustomBytes1,proto3" json:"CustomBytes1,omitempty"`
 	CustomBytes2   []byte `protobuf:"bytes,1012,opt,name=CustomBytes2,proto3" json:"CustomBytes2,omitempty"`
+	CustomBytes3   []byte `protobuf:"bytes,1013,opt,name=CustomBytes3,proto3" json:"CustomBytes3,omitempty"`
+	CustomBytes4   []byte `protobuf:"bytes,1014,opt,name=CustomBytes4,proto3" json:"CustomBytes4,omitempty"`
+	CustomBytes5   []byte `protobuf:"bytes,1015,opt,name=CustomBytes5,proto3" json:"CustomBytes5,omitempty"`
 }
 
 func (x *FlowMessage) Reset() {
@@ -243,6 +251,20 @@ func (x *FlowMessage) GetTimeFlowEnd() uint64 {
 	return 0
 }
 
+func (x *FlowMessage) GetTimeFlowStartMs() uint64 {
+	if x != nil {
+		return x.TimeFlowStartMs
+	}
+	return 0
+}
+
+func (x *FlowMessage) GetTimeFlowEndMs() uint64 {
+	if x != nil {
+		return x.TimeFlowEndMs
+	}
+	return 0
+}
+
 func (x *FlowMessage) GetBytes() uint64 {
 	if x != nil {
 		return x.Bytes
@@ -558,6 +580,27 @@ func (x *FlowMessage) GetCustomInteger2() uint64 {
 	return 0
 }
 
+func (x *FlowMessage) GetCustomInteger3() uint64 {
+	if x != nil {
+		return x.CustomInteger3
+	}
+	return 0
+}
+
+func (x *FlowMessage) GetCustomInteger4() uint64 {
+	if x != nil {
+		return x.CustomInteger4
+	}
+	return 0
+}
+
+func (x *FlowMessage) GetCustomInteger5() uint64 {
+	if x != nil {
+		return x.CustomInteger5
+	}
+	return 0
+}
+
 func (x *FlowMessage) GetCustomBytes1() []byte {
 	if x != nil {
 		return x.CustomBytes1
@@ -572,11 +615,32 @@ func (x *FlowMessage) GetCustomBytes2() []byte {
 	return nil
 }
 
+func (x *FlowMessage) GetCustomBytes3() []byte {
+	if x != nil {
+		return x.CustomBytes3
+	}
+	return nil
+}
+
+func (x *FlowMessage) GetCustomBytes4() []byte {
+	if x != nil {
+		return x.CustomBytes4
+	}
+	return nil
+}
+
+func (x *FlowMessage) GetCustomBytes5() []byte {
+	if x != nil {
+		return x.CustomBytes5
+	}
+	return nil
+}
+
 var File_pb_flow_proto protoreflect.FileDescriptor
 
 var file_pb_flow_proto_rawDesc = []byte{
 	0x0a, 0x0d, 0x70, 0x62, 0x2f, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12,
-	0x06, 0x66, 0x6c, 0x6f, 0x77, 0x70, 0x62, 0x22, 0xec, 0x0d, 0x0a, 0x0b, 0x46, 0x6c, 0x6f, 0x77,
+	0x06, 0x66, 0x6c, 0x6f, 0x77, 0x70, 0x62, 0x22, 0xa6, 0x10, 0x0a, 0x0b, 0x46, 0x6c, 0x6f, 0x77,
 	0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x30, 0x0a, 0x04, 0x54, 0x79, 0x70, 0x65, 0x18,
 	0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x1c, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x70, 0x62, 0x2e, 0x46,
 	0x6c, 0x6f, 0x77, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x2e, 0x46, 0x6c, 0x6f, 0x77, 0x54,
@@ -596,7 +660,12 @@ var file_pb_flow_proto_rawDesc = []byte{
 	0x72, 0x74, 0x18, 0x26, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0d, 0x54, 0x69, 0x6d, 0x65, 0x46, 0x6c,
 	0x6f, 0x77, 0x53, 0x74, 0x61, 0x72, 0x74, 0x12, 0x20, 0x0a, 0x0b, 0x54, 0x69, 0x6d, 0x65, 0x46,
 	0x6c, 0x6f, 0x77, 0x45, 0x6e, 0x64, 0x18, 0x05, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0b, 0x54, 0x69,
-	0x6d, 0x65, 0x46, 0x6c, 0x6f, 0x77, 0x45, 0x6e, 0x64, 0x12, 0x14, 0x0a, 0x05, 0x42, 0x79, 0x74,
+	0x6d, 0x65, 0x46, 0x6c, 0x6f, 0x77, 0x45, 0x6e, 0x64, 0x12, 0x28, 0x0a, 0x0f, 0x54, 0x69, 0x6d,
+	0x65, 0x46, 0x6c, 0x6f, 0x77, 0x53, 0x74, 0x61, 0x72, 0x74, 0x4d, 0x73, 0x18, 0x3f, 0x20, 0x01,
+	0x28, 0x04, 0x52, 0x0f, 0x54, 0x69, 0x6d, 0x65, 0x46, 0x6c, 0x6f, 0x77, 0x53, 0x74, 0x61, 0x72,
+	0x74, 0x4d, 0x73, 0x12, 0x24, 0x0a, 0x0d, 0x54, 0x69, 0x6d, 0x65, 0x46, 0x6c, 0x6f, 0x77, 0x45,
+	0x6e, 0x64, 0x4d, 0x73, 0x18, 0x40, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0d, 0x54, 0x69, 0x6d, 0x65,
+	0x46, 0x6c, 0x6f, 0x77, 0x45, 0x6e, 0x64, 0x4d, 0x73, 0x12, 0x14, 0x0a, 0x05, 0x42, 0x79, 0x74,
 	0x65, 0x73, 0x18, 0x09, 0x20, 0x01, 0x28, 0x04, 0x52, 0x05, 0x42, 0x79, 0x74, 0x65, 0x73, 0x12,
 	0x18, 0x0a, 0x07, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x73, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x04,
 	0x52, 0x07, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x73, 0x12, 0x18, 0x0a, 0x07, 0x53, 0x72, 0x63,
@@ -677,20 +746,35 @@ var file_pb_flow_proto_rawDesc = []byte{
 	0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x31, 0x12, 0x27,
 	0x0a, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x32,
 	0x18, 0xea, 0x07, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x49,
-	0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x32, 0x12, 0x23, 0x0a, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f,
+	0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x32, 0x12, 0x27, 0x0a, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f,
-	0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x31, 0x18, 0xf3, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0c,
+	0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x33, 0x18, 0xeb, 0x07, 0x20, 0x01, 0x28, 0x04,
-	0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x31, 0x12, 0x23, 0x0a, 0x0c,
+	0x52, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x33,
-	0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x32, 0x18, 0xf4, 0x07, 0x20,
+	0x12, 0x27, 0x0a, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65,
+	0x72, 0x34, 0x18, 0xec, 0x07, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f,
+	0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x34, 0x12, 0x27, 0x0a, 0x0e, 0x43, 0x75, 0x73,
+	0x74, 0x6f, 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x35, 0x18, 0xed, 0x07, 0x20, 0x01,
+	0x28, 0x04, 0x52, 0x0e, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65,
+	0x72, 0x35, 0x12, 0x23, 0x0a, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65,
+	0x73, 0x31, 0x18, 0xf3, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f,
+	0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x31, 0x12, 0x23, 0x0a, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f,
+	0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x32, 0x18, 0xf4, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0c,
+	0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x32, 0x12, 0x23, 0x0a, 0x0c,
+	0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x33, 0x18, 0xf5, 0x07, 0x20,
 	0x01, 0x28, 0x0c, 0x52, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73,
-	0x32, 0x22, 0x53, 0x0a, 0x08, 0x46, 0x6c, 0x6f, 0x77, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0f, 0x0a,
+	0x33, 0x12, 0x23, 0x0a, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73,
-	0x0b, 0x46, 0x4c, 0x4f, 0x57, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, 0x0b,
+	0x34, 0x18, 0xf6, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d,
-	0x0a, 0x07, 0x53, 0x46, 0x4c, 0x4f, 0x57, 0x5f, 0x35, 0x10, 0x01, 0x12, 0x0e, 0x0a, 0x0a, 0x4e,
+	0x42, 0x79, 0x74, 0x65, 0x73, 0x34, 0x12, 0x23, 0x0a, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d,
-	0x45, 0x54, 0x46, 0x4c, 0x4f, 0x57, 0x5f, 0x56, 0x35, 0x10, 0x02, 0x12, 0x0e, 0x0a, 0x0a, 0x4e,
+	0x42, 0x79, 0x74, 0x65, 0x73, 0x35, 0x18, 0xf7, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0c, 0x43,
-	0x45, 0x54, 0x46, 0x4c, 0x4f, 0x57, 0x5f, 0x56, 0x39, 0x10, 0x03, 0x12, 0x09, 0x0a, 0x05, 0x49,
+	0x75, 0x73, 0x74, 0x6f, 0x6d, 0x42, 0x79, 0x74, 0x65, 0x73, 0x35, 0x22, 0x53, 0x0a, 0x08, 0x46,
-	0x50, 0x46, 0x49, 0x58, 0x10, 0x04, 0x42, 0x29, 0x5a, 0x27, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62,
+	0x6c, 0x6f, 0x77, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0f, 0x0a, 0x0b, 0x46, 0x4c, 0x4f, 0x57, 0x55,
-	0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x6e, 0x65, 0x74, 0x73, 0x61, 0x6d, 0x70, 0x6c, 0x65, 0x72, 0x2f,
+	0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, 0x0b, 0x0a, 0x07, 0x53, 0x46, 0x4c, 0x4f,
-	0x67, 0x6f, 0x66, 0x6c, 0x6f, 0x77, 0x32, 0x2f, 0x70, 0x62, 0x3b, 0x66, 0x6c, 0x6f, 0x77, 0x70,
+	0x57, 0x5f, 0x35, 0x10, 0x01, 0x12, 0x0e, 0x0a, 0x0a, 0x4e, 0x45, 0x54, 0x46, 0x4c, 0x4f, 0x57,
-	0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x5f, 0x56, 0x35, 0x10, 0x02, 0x12, 0x0e, 0x0a, 0x0a, 0x4e, 0x45, 0x54, 0x46, 0x4c, 0x4f, 0x57,
+	0x5f, 0x56, 0x39, 0x10, 0x03, 0x12, 0x09, 0x0a, 0x05, 0x49, 0x50, 0x46, 0x49, 0x58, 0x10, 0x04,
+	0x42, 0x29, 0x5a, 0x27, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x6e,
+	0x65, 0x74, 0x73, 0x61, 0x6d, 0x70, 0x6c, 0x65, 0x72, 0x2f, 0x67, 0x6f, 0x66, 0x6c, 0x6f, 0x77,
+	0x32, 0x2f, 0x70, 0x62, 0x3b, 0x66, 0x6c, 0x6f, 0x77, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f,
+	0x74, 0x6f, 0x33,
 }
 
 var (

pb/flow.proto

@@ -25,6 +25,8 @@ message FlowMessage {
   // Found inside packet
   uint64 TimeFlowStart = 38;
   uint64 TimeFlowEnd = 5;
+  uint64 TimeFlowStartMs = 63;
+  uint64 TimeFlowEndMs = 64;
 
   // Size of the sampled packet
   uint64 Bytes = 9;

producer/producer_nf.go

@@ -340,46 +340,58 @@ func ConvertNetFlowDataSet(version uint16, baseTime uint32, uptime uint32, recor
 				case netflow.NFV9_FIELD_FIRST_SWITCHED:
 					var timeFirstSwitched uint32
 					DecodeUNumber(v, &timeFirstSwitched)
-					timeDiff := (uptime - timeFirstSwitched) / 1000
+					timeDiff := (uptime - timeFirstSwitched)
-					flowMessage.TimeFlowStart = uint64(baseTime - timeDiff)
+					flowMessage.TimeFlowStart = uint64(baseTime - timeDiff / 1000)
+					flowMessage.TimeFlowStartMs = uint64(baseTime)*1000 - uint64(timeDiff)
 				case netflow.NFV9_FIELD_LAST_SWITCHED:
 					var timeLastSwitched uint32
 					DecodeUNumber(v, &timeLastSwitched)
-					timeDiff := (uptime - timeLastSwitched) / 1000
+					timeDiff := (uptime - timeLastSwitched)
-					flowMessage.TimeFlowEnd = uint64(baseTime - timeDiff)
+					flowMessage.TimeFlowEnd = uint64(baseTime - timeDiff / 1000)
+					flowMessage.TimeFlowEndMs = uint64(baseTime)*1000 - uint64(timeDiff)
 				}
 			} else if version == 10 {
 				switch df.Type {
 				case netflow.IPFIX_FIELD_flowStartSeconds:
 					DecodeUNumber(v, &time)
 					flowMessage.TimeFlowStart = time
+					flowMessage.TimeFlowStartMs = time*1000
 				case netflow.IPFIX_FIELD_flowStartMilliseconds:
 					DecodeUNumber(v, &time)
 					flowMessage.TimeFlowStart = time / 1000
+					flowMessage.TimeFlowStartMs = time
 				case netflow.IPFIX_FIELD_flowStartMicroseconds:
 					DecodeUNumber(v, &time)
 					flowMessage.TimeFlowStart = time / 1000000
+					flowMessage.TimeFlowStartMs = time / 1000
 				case netflow.IPFIX_FIELD_flowStartNanoseconds:
 					DecodeUNumber(v, &time)
 					flowMessage.TimeFlowStart = time / 1000000000
+					flowMessage.TimeFlowStartMs = time / 1000000
 				case netflow.IPFIX_FIELD_flowEndSeconds:
 					DecodeUNumber(v, &time)
 					flowMessage.TimeFlowEnd = time
+					flowMessage.TimeFlowEndMs = time*1000
 				case netflow.IPFIX_FIELD_flowEndMilliseconds:
 					DecodeUNumber(v, &time)
 					flowMessage.TimeFlowEnd = time / 1000
+					flowMessage.TimeFlowEndMs = time
 				case netflow.IPFIX_FIELD_flowEndMicroseconds:
 					DecodeUNumber(v, &time)
 					flowMessage.TimeFlowEnd = time / 1000000
+					flowMessage.TimeFlowEndMs = time / 1000
 				case netflow.IPFIX_FIELD_flowEndNanoseconds:
 					DecodeUNumber(v, &time)
 					flowMessage.TimeFlowEnd = time / 1000000000
+					flowMessage.TimeFlowEndMs = time / 1000000
 				case netflow.IPFIX_FIELD_flowStartDeltaMicroseconds:
 					DecodeUNumber(v, &time)
 					flowMessage.TimeFlowStart = uint64(baseTime) - time/1000000
+					flowMessage.TimeFlowStartMs = uint64(baseTime) - time/1000
 				case netflow.IPFIX_FIELD_flowEndDeltaMicroseconds:
 					DecodeUNumber(v, &time)
 					flowMessage.TimeFlowEnd = uint64(baseTime) - time/1000000
+					flowMessage.TimeFlowEndMs = uint64(baseTime) - time/1000
 				// RFC7133
 				case netflow.IPFIX_FIELD_dataLinkFrameSize:
 					DecodeUNumber(v, &(flowMessage.Bytes))

producer/producer_nflegacy.go

@@ -14,10 +14,12 @@ func ConvertNetFlowLegacyRecord(baseTime uint32, uptime uint32, record netflowle
 
 	flowMessage.Type = flowmessage.FlowMessage_NETFLOW_V5
 
-	timeDiffFirst := (uptime - record.First) / 1000
+	timeDiffFirst := (uptime - record.First)
-	timeDiffLast := (uptime - record.Last) / 1000
+	timeDiffLast := (uptime - record.Last)
-	flowMessage.TimeFlowStart = uint64(baseTime - timeDiffFirst)
+	flowMessage.TimeFlowStart = uint64(baseTime - timeDiffFirst / 1000)
-	flowMessage.TimeFlowEnd = uint64(baseTime - timeDiffLast)
+	flowMessage.TimeFlowStartMs = uint64(baseTime)*1000 - uint64(timeDiffFirst)
+	flowMessage.TimeFlowEnd = uint64(baseTime - timeDiffLast / 1000)
+	flowMessage.TimeFlowEndMs = uint64(baseTime)*1000 - uint64(timeDiffLast)
 
 	v := make(net.IP, 4)
 	binary.BigEndian.PutUint32(v, record.NextHop)