NETOBSERV-934: Add icmpv4/v6 bits to FLP

go.mod

@@ -15,7 +15,7 @@ require (
 	github.com/mitchellh/mapstructure v1.4.3
 	github.com/netobserv/gopipes v0.3.0
 	github.com/netobserv/loki-client-go v0.0.0-20220927092034-f37122a54500
-	github.com/netobserv/netobserv-ebpf-agent v0.2.4
+	github.com/netobserv/netobserv-ebpf-agent v0.3.1-0.20230320150131-c62173ac9558
 	github.com/netsampler/goflow2 v1.1.1-0.20220509155230-5300494e4785
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.12.1

pkg/pipeline/decode/decode_protobuf.go

@@ -53,6 +53,8 @@ func PBFlowToMap(flow *pbflow.Record) config.GenericMap {
 		"Interface":       flow.Interface,
 		"AgentIP":         ipToStr(flow.AgentIp),
 		"Flags":           flow.Flags,
+		"IcmpType":        flow.Icmp.GetIcmpType(),
+		"IcmpCode":        flow.Icmp.GetIcmpCode(),
 	}
 	return out
 }

pkg/pipeline/decode/decode_protobuf_test.go

@@ -45,6 +45,10 @@ func TestDecodeProtobuf(t *testing.T) {
 			IpFamily: &pbflow.IP_Ipv4{Ipv4: 0x0a090807},
 		},
 		Flags: 0x100,
+		Icmp: &pbflow.Icmp{
+			IcmpType: 8,
+			IcmpCode: 0,
+		},
 	}
 	rawPB, err := proto.Marshal(&flow)
 	require.NoError(t, err)
@@ -71,6 +75,8 @@ func TestDecodeProtobuf(t *testing.T) {
 		"Interface":       "eth0",
 		"AgentIP":         "10.9.8.7",
 		"Flags":           uint32(0x100),
+		"IcmpType":        uint32(8),
+		"IcmpCode":        uint32(0),
 	}, out)
 }
 
@@ -106,6 +112,10 @@ func TestPBFlowToMap(t *testing.T) {
 			IpFamily: &pbflow.IP_Ipv4{Ipv4: 0x0a090807},
 		},
 		Flags: 0x100,
+		Icmp: &pbflow.Icmp{
+			IcmpType: 10,
+			IcmpCode: 11,
+		},
 	}
 
 	out := PBFlowToMap(flow)
@@ -129,6 +139,8 @@ func TestPBFlowToMap(t *testing.T) {
 		"Interface":       "eth0",
 		"AgentIP":         "10.9.8.7",
 		"Flags":           uint32(0x100),
+		"IcmpType":        uint32(10),
+		"IcmpCode":        uint32(11),
 	}, out)
 
 }

pkg/pipeline/pipeline_test.go

@@ -167,7 +167,7 @@ parameters:
 	// yield thread to allow pipe services correctly start
 	time.Sleep(10 * time.Millisecond)
 
-	flowSender, err := grpc.ConnectClient(fmt.Sprintf("127.0.0.1:%d", port))
+	flowSender, err := grpc.ConnectClient("127.0.0.1", port)
 	require.NoError(t, err)
 	defer flowSender.Close()
 
@@ -203,6 +203,10 @@ parameters:
 				IpFamily: &pbflow.IP_Ipv4{Ipv4: 0x0a0b0c0d},
 			},
 			Flags: 0x100,
+			Icmp: &pbflow.Icmp{
+				IcmpType: 10,
+				IcmpCode: 11,
+			},
 		}},
 	})
 	require.NoError(t, err)
@@ -233,6 +237,8 @@ parameters:
 		"Interface":       "eth0",
 		"AgentIP":         "10.11.12.13",
 		"Flags":           float64(0x100),
+		"IcmpType":        float64(10),
+		"IcmpCode":        float64(11),
 	}, capturedRecord)
 }