diff --git a/pb/flow.pb.go b/pb/flow.pb.go
index f1d6609..6194e3d 100644
--- a/pb/flow.pb.go
+++ b/pb/flow.pb.go
@@ -117,6 +117,7 @@ type FlowMessage struct {
 	IpTos            uint32 `protobuf:"varint,23,opt,name=ip_tos,json=ipTos,proto3" json:"ip_tos,omitempty"`
 	ForwardingStatus uint32 `protobuf:"varint,24,opt,name=forwarding_status,json=forwardingStatus,proto3" json:"forwarding_status,omitempty"`
 	IpTtl            uint32 `protobuf:"varint,25,opt,name=ip_ttl,json=ipTtl,proto3" json:"ip_ttl,omitempty"`
+	IpFlags          uint32 `protobuf:"varint,38,opt,name=ip_flags,json=ipFlags,proto3" json:"ip_flags,omitempty"`
 	TcpFlags         uint32 `protobuf:"varint,26,opt,name=tcp_flags,json=tcpFlags,proto3" json:"tcp_flags,omitempty"`
 	IcmpType         uint32 `protobuf:"varint,31,opt,name=icmp_type,json=icmpType,proto3" json:"icmp_type,omitempty"`
 	IcmpCode         uint32 `protobuf:"varint,32,opt,name=icmp_code,json=icmpCode,proto3" json:"icmp_code,omitempty"`
@@ -351,6 +352,13 @@ func (x *FlowMessage) GetIpTtl() uint32 {
 	return 0
 }
 
+func (x *FlowMessage) GetIpFlags() uint32 {
+	if x != nil {
+		return x.IpFlags
+	}
+	return 0
+}
+
 func (x *FlowMessage) GetTcpFlags() uint32 {
 	if x != nil {
 		return x.TcpFlags
@@ -495,7 +503,7 @@ 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, 0xd6, 0x0b, 0x0a, 0x0b, 0x46, 0x6c, 0x6f, 0x77,
+	0x06, 0x66, 0x6c, 0x6f, 0x77, 0x70, 0x62, 0x22, 0xf1, 0x0b, 0x0a, 0x0b, 0x46, 0x6c, 0x6f, 0x77,
 	0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x30, 0x0a, 0x04, 0x74, 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,
@@ -543,56 +551,57 @@ var file_pb_flow_proto_rawDesc = []byte{
 	0x64, 0x69, 0x6e, 0x67, 0x5f, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x18, 0x18, 0x20, 0x01, 0x28,
 	0x0d, 0x52, 0x10, 0x66, 0x6f, 0x72, 0x77, 0x61, 0x72, 0x64, 0x69, 0x6e, 0x67, 0x53, 0x74, 0x61,
 	0x74, 0x75, 0x73, 0x12, 0x15, 0x0a, 0x06, 0x69, 0x70, 0x5f, 0x74, 0x74, 0x6c, 0x18, 0x19, 0x20,
-	0x01, 0x28, 0x0d, 0x52, 0x05, 0x69, 0x70, 0x54, 0x74, 0x6c, 0x12, 0x1b, 0x0a, 0x09, 0x74, 0x63,
-	0x70, 0x5f, 0x66, 0x6c, 0x61, 0x67, 0x73, 0x18, 0x1a, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x74,
-	0x63, 0x70, 0x46, 0x6c, 0x61, 0x67, 0x73, 0x12, 0x1b, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70, 0x5f,
-	0x74, 0x79, 0x70, 0x65, 0x18, 0x1f, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x69, 0x63, 0x6d, 0x70,
-	0x54, 0x79, 0x70, 0x65, 0x12, 0x1b, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70, 0x5f, 0x63, 0x6f, 0x64,
-	0x65, 0x18, 0x20, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x69, 0x63, 0x6d, 0x70, 0x43, 0x6f, 0x64,
-	0x65, 0x12, 0x26, 0x0a, 0x0f, 0x69, 0x70, 0x76, 0x36, 0x5f, 0x66, 0x6c, 0x6f, 0x77, 0x5f, 0x6c,
-	0x61, 0x62, 0x65, 0x6c, 0x18, 0x25, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0d, 0x69, 0x70, 0x76, 0x36,
-	0x46, 0x6c, 0x6f, 0x77, 0x4c, 0x61, 0x62, 0x65, 0x6c, 0x12, 0x1f, 0x0a, 0x0b, 0x66, 0x72, 0x61,
-	0x67, 0x6d, 0x65, 0x6e, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x23, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0a,
-	0x66, 0x72, 0x61, 0x67, 0x6d, 0x65, 0x6e, 0x74, 0x49, 0x64, 0x12, 0x27, 0x0a, 0x0f, 0x66, 0x72,
-	0x61, 0x67, 0x6d, 0x65, 0x6e, 0x74, 0x5f, 0x6f, 0x66, 0x66, 0x73, 0x65, 0x74, 0x18, 0x24, 0x20,
-	0x01, 0x28, 0x0d, 0x52, 0x0e, 0x66, 0x72, 0x61, 0x67, 0x6d, 0x65, 0x6e, 0x74, 0x4f, 0x66, 0x66,
-	0x73, 0x65, 0x74, 0x12, 0x15, 0x0a, 0x06, 0x73, 0x72, 0x63, 0x5f, 0x61, 0x73, 0x18, 0x0e, 0x20,
-	0x01, 0x28, 0x0d, 0x52, 0x05, 0x73, 0x72, 0x63, 0x41, 0x73, 0x12, 0x15, 0x0a, 0x06, 0x64, 0x73,
-	0x74, 0x5f, 0x61, 0x73, 0x18, 0x0f, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x05, 0x64, 0x73, 0x74, 0x41,
-	0x73, 0x12, 0x19, 0x0a, 0x08, 0x6e, 0x65, 0x78, 0x74, 0x5f, 0x68, 0x6f, 0x70, 0x18, 0x0c, 0x20,
-	0x01, 0x28, 0x0c, 0x52, 0x07, 0x6e, 0x65, 0x78, 0x74, 0x48, 0x6f, 0x70, 0x12, 0x1e, 0x0a, 0x0b,
-	0x6e, 0x65, 0x78, 0x74, 0x5f, 0x68, 0x6f, 0x70, 0x5f, 0x61, 0x73, 0x18, 0x0d, 0x20, 0x01, 0x28,
-	0x0d, 0x52, 0x09, 0x6e, 0x65, 0x78, 0x74, 0x48, 0x6f, 0x70, 0x41, 0x73, 0x12, 0x17, 0x0a, 0x07,
-	0x73, 0x72, 0x63, 0x5f, 0x6e, 0x65, 0x74, 0x18, 0x10, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x06, 0x73,
-	0x72, 0x63, 0x4e, 0x65, 0x74, 0x12, 0x17, 0x0a, 0x07, 0x64, 0x73, 0x74, 0x5f, 0x6e, 0x65, 0x74,
-	0x18, 0x11, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x06, 0x64, 0x73, 0x74, 0x4e, 0x65, 0x74, 0x12, 0x20,
-	0x0a, 0x0c, 0x62, 0x67, 0x70, 0x5f, 0x6e, 0x65, 0x78, 0x74, 0x5f, 0x68, 0x6f, 0x70, 0x18, 0x64,
-	0x20, 0x01, 0x28, 0x0c, 0x52, 0x0a, 0x62, 0x67, 0x70, 0x4e, 0x65, 0x78, 0x74, 0x48, 0x6f, 0x70,
-	0x12, 0x27, 0x0a, 0x0f, 0x62, 0x67, 0x70, 0x5f, 0x63, 0x6f, 0x6d, 0x6d, 0x75, 0x6e, 0x69, 0x74,
-	0x69, 0x65, 0x73, 0x18, 0x65, 0x20, 0x03, 0x28, 0x0d, 0x52, 0x0e, 0x62, 0x67, 0x70, 0x43, 0x6f,
-	0x6d, 0x6d, 0x75, 0x6e, 0x69, 0x74, 0x69, 0x65, 0x73, 0x12, 0x17, 0x0a, 0x07, 0x61, 0x73, 0x5f,
-	0x70, 0x61, 0x74, 0x68, 0x18, 0x66, 0x20, 0x03, 0x28, 0x0d, 0x52, 0x06, 0x61, 0x73, 0x50, 0x61,
-	0x74, 0x68, 0x12, 0x19, 0x0a, 0x08, 0x6d, 0x70, 0x6c, 0x73, 0x5f, 0x74, 0x74, 0x6c, 0x18, 0x50,
-	0x20, 0x03, 0x28, 0x0d, 0x52, 0x07, 0x6d, 0x70, 0x6c, 0x73, 0x54, 0x74, 0x6c, 0x12, 0x1d, 0x0a,
-	0x0a, 0x6d, 0x70, 0x6c, 0x73, 0x5f, 0x6c, 0x61, 0x62, 0x65, 0x6c, 0x18, 0x51, 0x20, 0x03, 0x28,
-	0x0d, 0x52, 0x09, 0x6d, 0x70, 0x6c, 0x73, 0x4c, 0x61, 0x62, 0x65, 0x6c, 0x12, 0x17, 0x0a, 0x07,
-	0x6d, 0x70, 0x6c, 0x73, 0x5f, 0x69, 0x70, 0x18, 0x52, 0x20, 0x03, 0x28, 0x0c, 0x52, 0x06, 0x6d,
-	0x70, 0x6c, 0x73, 0x49, 0x70, 0x12, 0x32, 0x0a, 0x15, 0x6f, 0x62, 0x73, 0x65, 0x72, 0x76, 0x61,
-	0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x64, 0x6f, 0x6d, 0x61, 0x69, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x46,
-	0x20, 0x01, 0x28, 0x0d, 0x52, 0x13, 0x6f, 0x62, 0x73, 0x65, 0x72, 0x76, 0x61, 0x74, 0x69, 0x6f,
-	0x6e, 0x44, 0x6f, 0x6d, 0x61, 0x69, 0x6e, 0x49, 0x64, 0x12, 0x30, 0x0a, 0x14, 0x6f, 0x62, 0x73,
-	0x65, 0x72, 0x76, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x5f, 0x69,
-	0x64, 0x18, 0x47, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x12, 0x6f, 0x62, 0x73, 0x65, 0x72, 0x76, 0x61,
-	0x74, 0x69, 0x6f, 0x6e, 0x50, 0x6f, 0x69, 0x6e, 0x74, 0x49, 0x64, 0x22, 0x53, 0x0a, 0x08, 0x46,
-	0x6c, 0x6f, 0x77, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0f, 0x0a, 0x0b, 0x46, 0x4c, 0x4f, 0x57, 0x55,
-	0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, 0x0b, 0x0a, 0x07, 0x53, 0x46, 0x4c, 0x4f,
-	0x57, 0x5f, 0x35, 0x10, 0x01, 0x12, 0x0e, 0x0a, 0x0a, 0x4e, 0x45, 0x54, 0x46, 0x4c, 0x4f, 0x57,
-	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,
+	0x01, 0x28, 0x0d, 0x52, 0x05, 0x69, 0x70, 0x54, 0x74, 0x6c, 0x12, 0x19, 0x0a, 0x08, 0x69, 0x70,
+	0x5f, 0x66, 0x6c, 0x61, 0x67, 0x73, 0x18, 0x26, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x07, 0x69, 0x70,
+	0x46, 0x6c, 0x61, 0x67, 0x73, 0x12, 0x1b, 0x0a, 0x09, 0x74, 0x63, 0x70, 0x5f, 0x66, 0x6c, 0x61,
+	0x67, 0x73, 0x18, 0x1a, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x74, 0x63, 0x70, 0x46, 0x6c, 0x61,
+	0x67, 0x73, 0x12, 0x1b, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18,
+	0x1f, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x69, 0x63, 0x6d, 0x70, 0x54, 0x79, 0x70, 0x65, 0x12,
+	0x1b, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70, 0x5f, 0x63, 0x6f, 0x64, 0x65, 0x18, 0x20, 0x20, 0x01,
+	0x28, 0x0d, 0x52, 0x08, 0x69, 0x63, 0x6d, 0x70, 0x43, 0x6f, 0x64, 0x65, 0x12, 0x26, 0x0a, 0x0f,
+	0x69, 0x70, 0x76, 0x36, 0x5f, 0x66, 0x6c, 0x6f, 0x77, 0x5f, 0x6c, 0x61, 0x62, 0x65, 0x6c, 0x18,
+	0x25, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0d, 0x69, 0x70, 0x76, 0x36, 0x46, 0x6c, 0x6f, 0x77, 0x4c,
+	0x61, 0x62, 0x65, 0x6c, 0x12, 0x1f, 0x0a, 0x0b, 0x66, 0x72, 0x61, 0x67, 0x6d, 0x65, 0x6e, 0x74,
+	0x5f, 0x69, 0x64, 0x18, 0x23, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0a, 0x66, 0x72, 0x61, 0x67, 0x6d,
+	0x65, 0x6e, 0x74, 0x49, 0x64, 0x12, 0x27, 0x0a, 0x0f, 0x66, 0x72, 0x61, 0x67, 0x6d, 0x65, 0x6e,
+	0x74, 0x5f, 0x6f, 0x66, 0x66, 0x73, 0x65, 0x74, 0x18, 0x24, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0e,
+	0x66, 0x72, 0x61, 0x67, 0x6d, 0x65, 0x6e, 0x74, 0x4f, 0x66, 0x66, 0x73, 0x65, 0x74, 0x12, 0x15,
+	0x0a, 0x06, 0x73, 0x72, 0x63, 0x5f, 0x61, 0x73, 0x18, 0x0e, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x05,
+	0x73, 0x72, 0x63, 0x41, 0x73, 0x12, 0x15, 0x0a, 0x06, 0x64, 0x73, 0x74, 0x5f, 0x61, 0x73, 0x18,
+	0x0f, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x05, 0x64, 0x73, 0x74, 0x41, 0x73, 0x12, 0x19, 0x0a, 0x08,
+	0x6e, 0x65, 0x78, 0x74, 0x5f, 0x68, 0x6f, 0x70, 0x18, 0x0c, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07,
+	0x6e, 0x65, 0x78, 0x74, 0x48, 0x6f, 0x70, 0x12, 0x1e, 0x0a, 0x0b, 0x6e, 0x65, 0x78, 0x74, 0x5f,
+	0x68, 0x6f, 0x70, 0x5f, 0x61, 0x73, 0x18, 0x0d, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x09, 0x6e, 0x65,
+	0x78, 0x74, 0x48, 0x6f, 0x70, 0x41, 0x73, 0x12, 0x17, 0x0a, 0x07, 0x73, 0x72, 0x63, 0x5f, 0x6e,
+	0x65, 0x74, 0x18, 0x10, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x06, 0x73, 0x72, 0x63, 0x4e, 0x65, 0x74,
+	0x12, 0x17, 0x0a, 0x07, 0x64, 0x73, 0x74, 0x5f, 0x6e, 0x65, 0x74, 0x18, 0x11, 0x20, 0x01, 0x28,
+	0x0d, 0x52, 0x06, 0x64, 0x73, 0x74, 0x4e, 0x65, 0x74, 0x12, 0x20, 0x0a, 0x0c, 0x62, 0x67, 0x70,
+	0x5f, 0x6e, 0x65, 0x78, 0x74, 0x5f, 0x68, 0x6f, 0x70, 0x18, 0x64, 0x20, 0x01, 0x28, 0x0c, 0x52,
+	0x0a, 0x62, 0x67, 0x70, 0x4e, 0x65, 0x78, 0x74, 0x48, 0x6f, 0x70, 0x12, 0x27, 0x0a, 0x0f, 0x62,
+	0x67, 0x70, 0x5f, 0x63, 0x6f, 0x6d, 0x6d, 0x75, 0x6e, 0x69, 0x74, 0x69, 0x65, 0x73, 0x18, 0x65,
+	0x20, 0x03, 0x28, 0x0d, 0x52, 0x0e, 0x62, 0x67, 0x70, 0x43, 0x6f, 0x6d, 0x6d, 0x75, 0x6e, 0x69,
+	0x74, 0x69, 0x65, 0x73, 0x12, 0x17, 0x0a, 0x07, 0x61, 0x73, 0x5f, 0x70, 0x61, 0x74, 0x68, 0x18,
+	0x66, 0x20, 0x03, 0x28, 0x0d, 0x52, 0x06, 0x61, 0x73, 0x50, 0x61, 0x74, 0x68, 0x12, 0x19, 0x0a,
+	0x08, 0x6d, 0x70, 0x6c, 0x73, 0x5f, 0x74, 0x74, 0x6c, 0x18, 0x50, 0x20, 0x03, 0x28, 0x0d, 0x52,
+	0x07, 0x6d, 0x70, 0x6c, 0x73, 0x54, 0x74, 0x6c, 0x12, 0x1d, 0x0a, 0x0a, 0x6d, 0x70, 0x6c, 0x73,
+	0x5f, 0x6c, 0x61, 0x62, 0x65, 0x6c, 0x18, 0x51, 0x20, 0x03, 0x28, 0x0d, 0x52, 0x09, 0x6d, 0x70,
+	0x6c, 0x73, 0x4c, 0x61, 0x62, 0x65, 0x6c, 0x12, 0x17, 0x0a, 0x07, 0x6d, 0x70, 0x6c, 0x73, 0x5f,
+	0x69, 0x70, 0x18, 0x52, 0x20, 0x03, 0x28, 0x0c, 0x52, 0x06, 0x6d, 0x70, 0x6c, 0x73, 0x49, 0x70,
+	0x12, 0x32, 0x0a, 0x15, 0x6f, 0x62, 0x73, 0x65, 0x72, 0x76, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x5f,
+	0x64, 0x6f, 0x6d, 0x61, 0x69, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x46, 0x20, 0x01, 0x28, 0x0d, 0x52,
+	0x13, 0x6f, 0x62, 0x73, 0x65, 0x72, 0x76, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x44, 0x6f, 0x6d, 0x61,
+	0x69, 0x6e, 0x49, 0x64, 0x12, 0x30, 0x0a, 0x14, 0x6f, 0x62, 0x73, 0x65, 0x72, 0x76, 0x61, 0x74,
+	0x69, 0x6f, 0x6e, 0x5f, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x47, 0x20, 0x01,
+	0x28, 0x0d, 0x52, 0x12, 0x6f, 0x62, 0x73, 0x65, 0x72, 0x76, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x50,
+	0x6f, 0x69, 0x6e, 0x74, 0x49, 0x64, 0x22, 0x53, 0x0a, 0x08, 0x46, 0x6c, 0x6f, 0x77, 0x54, 0x79,
+	0x70, 0x65, 0x12, 0x0f, 0x0a, 0x0b, 0x46, 0x4c, 0x4f, 0x57, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57,
+	0x4e, 0x10, 0x00, 0x12, 0x0b, 0x0a, 0x07, 0x53, 0x46, 0x4c, 0x4f, 0x57, 0x5f, 0x35, 0x10, 0x01,
+	0x12, 0x0e, 0x0a, 0x0a, 0x4e, 0x45, 0x54, 0x46, 0x4c, 0x4f, 0x57, 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 (
diff --git a/pb/flow.proto b/pb/flow.proto
index 25fa14a..ef72649 100644
--- a/pb/flow.proto
+++ b/pb/flow.proto
@@ -62,6 +62,7 @@ message FlowMessage {
   uint32 ip_tos = 23;
   uint32 forwarding_status = 24;
   uint32 ip_ttl = 25;
+  uint32 ip_flags = 38;
   uint32 tcp_flags = 26;
   uint32 icmp_type = 31;
   uint32 icmp_code = 32;
diff --git a/producer/proto/producer_nf.go b/producer/proto/producer_nf.go
index 81a4b97..0d3b382 100644
--- a/producer/proto/producer_nf.go
+++ b/producer/proto/producer_nf.go
@@ -485,13 +485,13 @@ func ConvertNetFlowDataSet(flowMessage *ProtoProducerMessage, version uint16, ba
 			if err := DecodeUNumber(v, &fragOffset); err != nil {
 				return err
 			}
-			flowMessage.FragmentOffset |= fragOffset
+			flowMessage.FragmentOffset = fragOffset
 		case netflow.IPFIX_FIELD_fragmentFlags:
 			var ipFlags uint32
 			if err := DecodeUNumber(v, &ipFlags); err != nil {
 				return err
 			}
-			flowMessage.FragmentOffset |= ipFlags
+			flowMessage.IpFlags = ipFlags
 		case netflow.NFV9_FIELD_IPV6_FLOW_LABEL:
 			if err := DecodeUNumber(v, &(flowMessage.Ipv6FlowLabel)); err != nil {
 				return err
diff --git a/producer/proto/producer_sf.go b/producer/proto/producer_sf.go
index 566c800..c06820a 100644
--- a/producer/proto/producer_sf.go
+++ b/producer/proto/producer_sf.go
@@ -25,76 +25,233 @@ func ParseSampledHeader(flowMessage *ProtoProducerMessage, sampledHeader *sflow.
 	return ParseSampledHeaderConfig(flowMessage, sampledHeader, nil)
 }
 
-func ParseEthernetHeader(flowMessage *ProtoProducerMessage, data []byte, config *SFlowMapper) error {
+func ParseEthernet(offset int, flowMessage *ProtoProducerMessage, data []byte) (etherType []byte, newOffset int, err error) {
+	if len(data) >= offset+14 {
+		etherType = data[offset+12 : offset+14]
+
+		dstMac := binary.BigEndian.Uint64(append([]byte{0, 0}, data[offset+0:offset+6]...))
+		srcMac := binary.BigEndian.Uint64(append([]byte{0, 0}, data[offset+6:offset+12]...))
+		flowMessage.SrcMac = srcMac
+		flowMessage.DstMac = dstMac
+
+		offset += 14
+	}
+	return etherType, offset, err
+}
+
+func Parse8021Q(offset int, flowMessage *ProtoProducerMessage, data []byte) (etherType []byte, newOffset int, err error) {
+	if len(data) >= offset+4 {
+		flowMessage.VlanId = uint32(binary.BigEndian.Uint16(data[offset : offset+2]))
+		etherType = data[offset+2 : offset+4]
+
+		offset += 4
+	}
+	return etherType, offset, err
+}
+
+func ParseMPLS(offset int, flowMessage *ProtoProducerMessage, data []byte) (etherType []byte, newOffset int, err error) {
 	var mplsLabel []uint32
 	var mplsTtl []uint32
 
+	iterateMpls := true
+	for iterateMpls {
+		if len(data) < offset+5 {
+			// stop iterating mpls, not enough payload left
+			break
+		}
+		label := binary.BigEndian.Uint32(append([]byte{0}, data[offset:offset+3]...)) >> 4
+		//exp := data[offset+2] > 1
+		bottom := data[offset+2] & 1
+		ttl := data[offset+3]
+		offset += 4
+
+		if bottom == 1 || label <= 15 || offset > len(data) {
+			if data[offset]&0xf0>>4 == 4 {
+				etherType = []byte{0x8, 0x0}
+			} else if data[offset]&0xf0>>4 == 6 {
+				etherType = []byte{0x86, 0xdd}
+			}
+			iterateMpls = false // stop iterating mpls, bottom of stack
+		}
+
+		mplsLabel = append(mplsLabel, label)
+		mplsTtl = append(mplsTtl, uint32(ttl))
+	}
+	// if multiple MPLS headers, will reset existing values
+	flowMessage.MplsLabel = mplsLabel
+	flowMessage.MplsTtl = mplsTtl
+	return etherType, offset, err
+}
+
+func ParseIPv4(offset int, flowMessage *ProtoProducerMessage, data []byte) (nextHeader byte, newOffset int, err error) {
+	if len(data) >= offset+20 {
+		nextHeader = data[offset+9]
+		flowMessage.SrcAddr = data[offset+12 : offset+16]
+		flowMessage.DstAddr = data[offset+16 : offset+20]
+
+		tos := data[offset+1]
+		ttl := data[offset+8]
+
+		flowMessage.IpTos = uint32(tos)
+		flowMessage.IpTtl = uint32(ttl)
+
+		identification := binary.BigEndian.Uint16(data[offset+4 : offset+6])
+		fragOffset := binary.BigEndian.Uint16(data[offset+6 : offset+8]) // also includes flag
+
+		flowMessage.FragmentId = uint32(identification)
+		flowMessage.FragmentOffset = uint32(fragOffset) & 57344
+		flowMessage.IpFlags = uint32(fragOffset) >> 13
+
+		offset += 20
+	}
+	return nextHeader, offset, err
+}
+
+func ParseIPv6(offset int, flowMessage *ProtoProducerMessage, data []byte) (nextHeader byte, newOffset int, err error) {
+	if len(data) >= offset+40 {
+		nextHeader = data[offset+6]
+		flowMessage.SrcAddr = data[offset+8 : offset+24]
+		flowMessage.DstAddr = data[offset+24 : offset+40]
+
+		tostmp := uint32(binary.BigEndian.Uint16(data[offset : offset+2]))
+		tos := uint8(tostmp & 0x0ff0 >> 4)
+		ttl := data[offset+7]
+
+		flowMessage.IpTos = uint32(tos)
+		flowMessage.IpTtl = uint32(ttl)
+
+		flowLabel := binary.BigEndian.Uint32(data[offset : offset+4])
+		flowMessage.Ipv6FlowLabel = flowLabel & 0xFFFFF
+
+		offset += 40
+	}
+	return nextHeader, offset, err
+}
+
+func ParseIPv6Headers(nextHeader byte, offset int, flowMessage *ProtoProducerMessage, data []byte) (newNextHeader byte, newOffset int, err error) {
+	for {
+		if nextHeader == 44 && len(data) >= offset+8 {
+			nextHeader = data[offset]
+
+			fragOffset := binary.BigEndian.Uint16(data[offset+2 : offset+4]) // also includes flag
+			identification := binary.BigEndian.Uint32(data[offset+4 : offset+8])
+
+			flowMessage.FragmentId = identification
+			flowMessage.FragmentOffset = uint32(fragOffset) >> 3
+			flowMessage.IpFlags = uint32(fragOffset) & 7
+
+			offset += 8
+		} else {
+			break
+		}
+	}
+	return nextHeader, offset, err
+}
+
+func ParseTCP(offset int, flowMessage *ProtoProducerMessage, data []byte) (newOffset int, err error) {
+	if len(data) >= offset+13 {
+		srcPort := binary.BigEndian.Uint16(data[offset+0 : offset+2])
+		dstPort := binary.BigEndian.Uint16(data[offset+2 : offset+4])
+
+		flowMessage.SrcPort = uint32(srcPort)
+		flowMessage.DstPort = uint32(dstPort)
+
+		tcpflags := data[offset+13]
+		flowMessage.TcpFlags = uint32(tcpflags)
+
+		length := int(data[13]>>4) * 4
+
+		offset += length
+	}
+	return offset, err
+}
+
+func ParseUDP(offset int, flowMessage *ProtoProducerMessage, data []byte) (newOffset int, err error) {
+	if len(data) >= offset+4 {
+		srcPort := binary.BigEndian.Uint16(data[offset+0 : offset+2])
+		dstPort := binary.BigEndian.Uint16(data[offset+2 : offset+4])
+
+		flowMessage.SrcPort = uint32(srcPort)
+		flowMessage.DstPort = uint32(dstPort)
+
+		offset += 8
+	}
+	return offset, err
+}
+
+func ParseICMP(offset int, flowMessage *ProtoProducerMessage, data []byte) (newOffset int, err error) {
+	if len(data) >= offset+2 {
+		flowMessage.IcmpType = uint32(data[offset+0])
+		flowMessage.IcmpCode = uint32(data[offset+1])
+
+		offset += 8
+	}
+	return offset, err
+}
+
+func ParseICMPv6(offset int, flowMessage *ProtoProducerMessage, data []byte) (newOffset int, err error) {
+	if len(data) >= offset+2 {
+		flowMessage.IcmpType = uint32(data[offset+0])
+		flowMessage.IcmpCode = uint32(data[offset+1])
+
+		offset += 8
+	}
+	return offset, err
+}
+
+func IsMPLS(etherType []byte) bool {
+	return etherType[0] == 0x88 && etherType[1] == 0x47
+}
+
+func Is8021Q(etherType []byte) bool {
+	return etherType[0] == 0x81 && etherType[1] == 0x0
+}
+
+func IsIPv4(etherType []byte) bool {
+	return etherType[0] == 0x8 && etherType[1] == 0x0
+}
+
+func IsIPv6(etherType []byte) bool {
+	return etherType[0] == 0x86 && etherType[1] == 0xdd
+}
+
+func IsARP(etherType []byte) bool {
+	return etherType[0] == 0x8 && etherType[1] == 0x6
+}
+
+// Parses an entire stream consisting of multiple layers of protocols
+// It picks the best field to map when multiple encapsulation of the same layer (eg: tunnels, extension headers, etc.)
+func ParseEthernetHeader(flowMessage *ProtoProducerMessage, data []byte, config *SFlowMapper) (err error) {
 	var nextHeader byte
-	var tcpflags byte
-	var srcIP, dstIP []byte
-	offset := 14
+	var offset int
 
-	var srcMac uint64
-	var dstMac uint64
-
-	var tos byte
-	var ttl byte
-	var identification uint16
-	var fragOffset uint16
-	var flowLabel uint32
-
-	var srcPort uint16
-	var dstPort uint16
+	var etherType []byte
 
 	for _, configLayer := range GetSFlowConfigLayer(config, "0") {
-		extracted := GetBytes(data, configLayer.Offset, configLayer.Length)
+		extracted := GetBytes(data, offset+configLayer.Offset, configLayer.Length)
 		if err := MapCustom(flowMessage, extracted, configLayer.MapConfigBase); err != nil {
 			return err
 		}
 	}
 
-	etherType := data[12:14]
-
-	dstMac = binary.BigEndian.Uint64(append([]byte{0, 0}, data[0:6]...))
-	srcMac = binary.BigEndian.Uint64(append([]byte{0, 0}, data[6:12]...))
-	flowMessage.SrcMac = srcMac
-	flowMessage.DstMac = dstMac
+	if etherType, offset, err = ParseEthernet(offset, flowMessage, data); err != nil {
+		return err
+	}
 
 	encap := true
 	iterations := 0
 	for encap && iterations <= 1 {
 		encap = false
 
-		if etherType[0] == 0x81 && etherType[1] == 0x0 { // VLAN 802.1Q
-			flowMessage.VlanId = uint32(binary.BigEndian.Uint16(data[14:16]))
-			offset += 4
-			etherType = data[16:18]
+		if Is8021Q(etherType) { // VLAN 802.1Q
+			if etherType, offset, err = Parse8021Q(offset, flowMessage, data); err != nil {
+				return err
+			}
 		}
 
-		if etherType[0] == 0x88 && etherType[1] == 0x47 { // MPLS
-			iterateMpls := true
-			for iterateMpls {
-				if len(data) < offset+5 {
-					// stop iterating mpls, not enough payload left
-					break
-				}
-				label := binary.BigEndian.Uint32(append([]byte{0}, data[offset:offset+3]...)) >> 4
-				//exp := data[offset+2] > 1
-				bottom := data[offset+2] & 1
-				ttl := data[offset+3]
-				offset += 4
-
-				if bottom == 1 || label <= 15 || offset > len(data) {
-					if data[offset]&0xf0>>4 == 4 {
-						etherType = []byte{0x8, 0x0}
-					} else if data[offset]&0xf0>>4 == 6 {
-						etherType = []byte{0x86, 0xdd}
-					}
-					iterateMpls = false // stop iterating mpls, bottom of stack
-				}
-
-				mplsLabel = append(mplsLabel, label)
-				mplsTtl = append(mplsTtl, uint32(ttl))
+		if IsMPLS(etherType) { // MPLS
+			if etherType, offset, err = ParseMPLS(offset, flowMessage, data); err != nil {
+				return err
 			}
 		}
 
@@ -105,49 +262,34 @@ func ParseEthernetHeader(flowMessage *ProtoProducerMessage, data []byte, config
 			}
 		}
 
-		if etherType[0] == 0x8 && etherType[1] == 0x0 { // IPv4
-			if len(data) >= offset+20 {
-				nextHeader = data[offset+9]
-				srcIP = data[offset+12 : offset+16]
-				dstIP = data[offset+16 : offset+20]
-				tos = data[offset+1]
-				ttl = data[offset+8]
-
-				identification = binary.BigEndian.Uint16(data[offset+4 : offset+6])
-				fragOffset = binary.BigEndian.Uint16(data[offset+6:offset+8]) & 8191
-
-				for _, configLayer := range GetSFlowConfigLayer(config, "ipv4") {
-					extracted := GetBytes(data, offset*8+configLayer.Offset, configLayer.Length)
-					if err := MapCustom(flowMessage, extracted, configLayer.MapConfigBase); err != nil {
-						return err
-					}
-				}
-
-				offset += 20
+		if IsIPv4(etherType) { // IPv4
+			prevOffset := offset
+			if nextHeader, offset, err = ParseIPv4(offset, flowMessage, data); err != nil {
+				return err
 			}
-		} else if etherType[0] == 0x86 && etherType[1] == 0xdd { // IPv6
-			if len(data) >= offset+40 {
-				nextHeader = data[offset+6]
-				srcIP = data[offset+8 : offset+24]
-				dstIP = data[offset+24 : offset+40]
 
-				tostmp := uint32(binary.BigEndian.Uint16(data[offset : offset+2]))
-				tos = uint8(tostmp & 0x0ff0 >> 4)
-				ttl = data[offset+7]
-
-				flowLabel = binary.BigEndian.Uint32(data[offset : offset+4])
-
-				for _, configLayer := range GetSFlowConfigLayer(config, "ipv6") {
-					extracted := GetBytes(data, offset*8+configLayer.Offset, configLayer.Length)
-					if err := MapCustom(flowMessage, extracted, configLayer.MapConfigBase); err != nil {
-						return err
-					}
+			for _, configLayer := range GetSFlowConfigLayer(config, "ipv4") {
+				extracted := GetBytes(data, prevOffset*8+configLayer.Offset, configLayer.Length)
+				if err := MapCustom(flowMessage, extracted, configLayer.MapConfigBase); err != nil {
+					return err
 				}
-
-				offset += 40
-
 			}
-		} else if etherType[0] == 0x8 && etherType[1] == 0x6 { // ARP
+		} else if IsIPv6(etherType) { // IPv6
+			prevOffset := offset
+			if nextHeader, offset, err = ParseIPv6(offset, flowMessage, data); err != nil {
+				return err
+			}
+			if nextHeader, offset, err = ParseIPv6Headers(nextHeader, offset, flowMessage, data); err != nil {
+				return err
+			}
+
+			for _, configLayer := range GetSFlowConfigLayer(config, "ipv6") {
+				extracted := GetBytes(data, prevOffset*8+configLayer.Offset, configLayer.Length)
+				if err := MapCustom(flowMessage, extracted, configLayer.MapConfigBase); err != nil {
+					return err
+				}
+			}
+		} else if IsARP(etherType) { // ARP
 			for _, configLayer := range GetSFlowConfigLayer(config, "arp") {
 				extracted := GetBytes(data, offset*8+configLayer.Offset, configLayer.Length)
 				if err := MapCustom(flowMessage, extracted, configLayer.MapConfigBase); err != nil {
@@ -163,61 +305,63 @@ func ParseEthernetHeader(flowMessage *ProtoProducerMessage, data []byte, config
 			}
 		}
 
-		appOffset := 0
-		if len(data) >= offset+4 && (nextHeader == 17 || nextHeader == 6) && fragOffset&8191 == 0 {
-			srcPort = binary.BigEndian.Uint16(data[offset+0 : offset+2])
-			dstPort = binary.BigEndian.Uint16(data[offset+2 : offset+4])
-		}
+		var appOffset int // keeps track of the user payload
 
-		if nextHeader == 17 {
-			appOffset = 8
-
-			for _, configLayer := range GetSFlowConfigLayer(config, "udp") {
-				extracted := GetBytes(data, offset*8+configLayer.Offset, configLayer.Length)
-				if err := MapCustom(flowMessage, extracted, configLayer.MapConfigBase); err != nil {
-					return err
-				}
-			}
-		}
-
-		if len(data) > offset+13 && nextHeader == 6 {
-			tcpflags = data[offset+13]
-
-			appOffset = int(data[13]>>4) * 4
-
-			for _, configLayer := range GetSFlowConfigLayer(config, "tcp") {
-				extracted := GetBytes(data, offset*8+configLayer.Offset, configLayer.Length)
-				if err := MapCustom(flowMessage, extracted, configLayer.MapConfigBase); err != nil {
-					return err
-				}
-			}
-		}
-
-		// ICMP and ICMPv6
-		if len(data) >= offset+2 && (nextHeader == 1 || nextHeader == 58) {
-			flowMessage.IcmpType = uint32(data[offset+0])
-			flowMessage.IcmpCode = uint32(data[offset+1])
-
-			if nextHeader == 1 {
-				for _, configLayer := range GetSFlowConfigLayer(config, "icmp") {
-					extracted := GetBytes(data, offset*8+configLayer.Offset, configLayer.Length)
-					if err := MapCustom(flowMessage, extracted, configLayer.MapConfigBase); err != nil {
+		// Transport protocols
+		if nextHeader == 17 || nextHeader == 6 || nextHeader == 1 || nextHeader == 58 {
+			prevOffset := offset
+			if flowMessage.FragmentOffset == 0 {
+				if nextHeader == 17 { // UDP
+					if offset, err = ParseUDP(offset, flowMessage, data); err != nil {
 						return err
 					}
-				}
-			} else if nextHeader == 58 {
-				for _, configLayer := range GetSFlowConfigLayer(config, "icmp6") {
-					extracted := GetBytes(data, offset*8+configLayer.Offset, configLayer.Length)
-					if err := MapCustom(flowMessage, extracted, configLayer.MapConfigBase); err != nil {
+					for _, configLayer := range GetSFlowConfigLayer(config, "udp") {
+						extracted := GetBytes(data, prevOffset*8+configLayer.Offset, configLayer.Length)
+						if err := MapCustom(flowMessage, extracted, configLayer.MapConfigBase); err != nil {
+							return err
+						}
+					}
+				} else if nextHeader == 6 { // TCP
+					if offset, err = ParseTCP(offset, flowMessage, data); err != nil {
 						return err
 					}
+					for _, configLayer := range GetSFlowConfigLayer(config, "tcp") {
+						extracted := GetBytes(data, prevOffset*8+configLayer.Offset, configLayer.Length)
+						if err := MapCustom(flowMessage, extracted, configLayer.MapConfigBase); err != nil {
+							return err
+						}
+					}
+				} else if nextHeader == 1 { // ICMP
+					if offset, err = ParseICMP(offset, flowMessage, data); err != nil {
+						return err
+					}
+					for _, configLayer := range GetSFlowConfigLayer(config, "icmp") {
+						extracted := GetBytes(data, prevOffset*8+configLayer.Offset, configLayer.Length)
+						if err := MapCustom(flowMessage, extracted, configLayer.MapConfigBase); err != nil {
+							return err
+						}
+					}
+				} else if nextHeader == 58 { // ICMPv6
+					if offset, err = ParseICMPv6(offset, flowMessage, data); err != nil {
+						return err
+					}
+					for _, configLayer := range GetSFlowConfigLayer(config, "icmp6") {
+						extracted := GetBytes(data, prevOffset*8+configLayer.Offset, configLayer.Length)
+						if err := MapCustom(flowMessage, extracted, configLayer.MapConfigBase); err != nil {
+							return err
+						}
+					}
 				}
 			}
+			appOffset = offset
 		}
 
+		// fetch data from the application/payload
 		if appOffset > 0 {
 			for _, configLayer := range GetSFlowConfigLayer(config, "7") {
-				extracted := GetBytes(data, (offset+appOffset)*8+configLayer.Offset, configLayer.Length)
+				customOffset := appOffset*8 + configLayer.Offset - int(flowMessage.FragmentOffset)*8 // allows user to get data from a fragment as well
+				// todo: check the calculation (might be off due to various header size)
+				extracted := GetBytes(data, customOffset, configLayer.Length)
 				if err := MapCustom(flowMessage, extracted, configLayer.MapConfigBase); err != nil {
 					return err
 				}
@@ -227,24 +371,8 @@ func ParseEthernetHeader(flowMessage *ProtoProducerMessage, data []byte, config
 		iterations++
 	}
 
-	flowMessage.MplsLabel = mplsLabel
-	flowMessage.MplsTtl = mplsTtl
-
 	flowMessage.Etype = uint32(binary.BigEndian.Uint16(etherType[0:2]))
-	flowMessage.Ipv6FlowLabel = flowLabel & 0xFFFFF
-
-	flowMessage.SrcPort = uint32(srcPort)
-	flowMessage.DstPort = uint32(dstPort)
-
-	flowMessage.SrcAddr = srcIP
-	flowMessage.DstAddr = dstIP
 	flowMessage.Proto = uint32(nextHeader)
-	flowMessage.IpTos = uint32(tos)
-	flowMessage.IpTtl = uint32(ttl)
-	flowMessage.TcpFlags = uint32(tcpflags)
-
-	flowMessage.FragmentId = uint32(identification)
-	flowMessage.FragmentOffset = uint32(fragOffset)
 
 	return nil
 }
@@ -283,7 +411,7 @@ func SearchSFlowSampleConfig(flowMessage *ProtoProducerMessage, flowSample inter
 		switch recordData := record.Data.(type) {
 		case sflow.SampledHeader:
 			flowMessage.Bytes = uint64(recordData.FrameLength)
-			if err := ParseSampledHeaderConfig(flowMessage, &recordData, config); err != nil {
+			if err := ParseSampledHeaderConfig(flowMessage, &recordData, config); err != nil { // todo: make function configurable
 				return err
 			}
 		case sflow.SampledIPv4:
diff --git a/producer/proto/producer_test.go b/producer/proto/producer_test.go
index 8d7cca5..57037fc 100644
--- a/producer/proto/producer_test.go
+++ b/producer/proto/producer_test.go
@@ -1,6 +1,8 @@
 package protoproducer
 
 import (
+	"encoding/hex"
+	"encoding/json"
 	"testing"
 
 	"github.com/netsampler/goflow2/v2/decoders/netflow"
@@ -191,3 +193,72 @@ func getSflowPacket() *sflow.Packet {
 	}
 	return &pkt
 }
+
+func TestProcessEthernet(t *testing.T) {
+	dataStr := "005300000001" + // src mac
+		"005300000002" + // dst mac
+		"86dd" + // etype
+		"6000000004d83a40" + // ipv6
+		"fd010000000000000000000000000001" + // src
+		"fd010000000000000000000000000002" + // dst
+		"8000f96508a4" // icmpv6
+	data, _ := hex.DecodeString(dataStr)
+	var flowMessage ProtoProducerMessage
+	err := ParseEthernetHeader(&flowMessage, data, nil)
+	assert.Nil(t, err)
+
+	b, _ := json.Marshal(flowMessage.FlowMessage)
+	t.Log(string(b))
+
+	assert.Equal(t, uint32(0x86dd), flowMessage.Etype)
+	assert.Equal(t, uint32(58), flowMessage.Proto)
+	assert.Equal(t, uint32(128), flowMessage.IcmpType)
+}
+
+func TestProcessIPv6Headers(t *testing.T) {
+	dataStr := "6000000004d82c40" +
+		"fd010000000000000000000000000001" + // src
+		"fd010000000000000000000000000002" + // dst
+		"3a000001a7882ea9" + // fragment header
+		"8000f96508a4" // icmpv6
+	data, _ := hex.DecodeString(dataStr)
+	var flowMessage ProtoProducerMessage
+	nextHeader, offset, err := ParseIPv6(0, &flowMessage, data)
+	assert.Nil(t, err)
+	assert.Equal(t, byte(44), nextHeader)
+	nextHeader, offset, err = ParseIPv6Headers(nextHeader, offset, &flowMessage, data)
+	assert.Nil(t, err)
+	assert.Equal(t, byte(58), nextHeader)
+
+	offset, err = ParseICMPv6(offset, &flowMessage, data)
+	assert.Nil(t, err)
+
+	b, _ := json.Marshal(flowMessage.FlowMessage)
+	t.Log(string(b), nextHeader, offset)
+
+	assert.Equal(t, uint32(1), flowMessage.IpFlags)
+	assert.Equal(t, uint32(64), flowMessage.IpTtl)
+	assert.Equal(t, uint32(2810719913), flowMessage.FragmentId)
+	assert.Equal(t, uint32(0), flowMessage.FragmentOffset)
+	assert.Equal(t, uint32(128), flowMessage.IcmpType)
+}
+
+func TestProcessIPv4Fragment(t *testing.T) {
+	dataStr := "450002245dd900b94001ffe1" +
+		"c0a80101" + // src
+		"c0a80102" + // dst
+		"0809" // continued payload
+	data, _ := hex.DecodeString(dataStr)
+	var flowMessage ProtoProducerMessage
+	nextHeader, offset, err := ParseIPv4(0, &flowMessage, data)
+	assert.Nil(t, err)
+	assert.Equal(t, byte(1), nextHeader)
+
+	b, _ := json.Marshal(flowMessage.FlowMessage)
+	t.Log(string(b), nextHeader, offset)
+
+	assert.Equal(t, uint32(0), flowMessage.IpFlags)
+	assert.Equal(t, uint32(64), flowMessage.IpTtl)
+	assert.Equal(t, uint32(24025), flowMessage.FragmentId)
+	assert.Equal(t, uint32(0), flowMessage.FragmentOffset)
+}
diff --git a/producer/proto/reflect.go b/producer/proto/reflect.go
index b8efc82..6874a70 100644
--- a/producer/proto/reflect.go
+++ b/producer/proto/reflect.go
@@ -27,7 +27,7 @@ var (
 )
 
 func GetBytes(d []byte, offset int, length int) []byte {
-	if length == 0 {
+	if length == 0 || offset < 0 {
 		return nil
 	}
 	leftBytes := offset / 8