inlet/flow: fix parsing of sFlow IPv4/IPv6 data

Fix #470

console/data/docs/99-changelog.md

@@ -17,6 +17,7 @@ When upgrading to this release, it takes some time to reduce the storage size
 for a few columns.
 
 - ✨ *console*: add “%” to available units
+- 🩹 *inlet*: fix parsing of sFlow IPv4/IPv6 data
 - 🩹 *orchestrator*: fix disabling of `DstASPath`
 - 🩹 *console*: fix time range selection
 - 🩹 *console*: fix calculation of the L2 overhead when selecting L2 bps

go.mod

@@ -29,6 +29,7 @@ require (
 	github.com/mattn/go-isatty v0.0.17
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/netsampler/goflow2 v1.1.1-0.20221008154147-57fad2e0c837
+	github.com/opencontainers/image-spec v1.0.3-0.20211202183452-c5a74bcca799
 	github.com/oschwald/maxminddb-golang v1.10.0
 	github.com/osrg/gobgp/v3 v3.11.0
 	github.com/prometheus/client_golang v1.14.0
@@ -97,7 +98,6 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
-	github.com/opencontainers/image-spec v1.0.3-0.20211202183452-c5a74bcca799 // indirect
 	github.com/paulmach/orb v0.8.0 // indirect
 	github.com/pelletier/go-toml/v2 v2.0.6 // indirect
 	github.com/pierrec/lz4/v4 v4.1.17 // indirect
@@ -137,5 +137,5 @@ require (
 replace (
 	github.com/benbjohnson/clock => github.com/vincentbernat/go-clock v0.0.0-20220922224448-739bd11b5833
 	github.com/kentik/patricia => github.com/vincentbernat/patricia v0.0.0-20220923091046-b376a1167a94
-	github.com/netsampler/goflow2 => github.com/vincentbernat/goflow2 v1.0.5-0.20230119111129-c797534c51e4
+	github.com/netsampler/goflow2 => github.com/vincentbernat/goflow2 v1.0.5-0.20230210215259-34e153ff515d
 )

go.sum

@@ -465,8 +465,8 @@ github.com/ugorji/go/codec v1.2.7/go.mod h1:WGN1fab3R1fzQlVQTkfxVtIBhWDRqOviHU95
 github.com/urfave/cli/v2 v2.1.1/go.mod h1:SE9GqnLQmjVa0iPEY0f1w3ygNIYcIJ0OKPMoW2caLfQ=
 github.com/vincentbernat/go-clock v0.0.0-20220922224448-739bd11b5833 h1:eeHgOFlrGNESR9TF+AJovNWOxH8AdmXWK2nGXKa6RUU=
 github.com/vincentbernat/go-clock v0.0.0-20220922224448-739bd11b5833/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
-github.com/vincentbernat/goflow2 v1.0.5-0.20230119111129-c797534c51e4 h1:peywq/qLmgBq11JZgus3XqnOf8NqDnSQ7ygi+X0O6w8=
-github.com/vincentbernat/goflow2 v1.0.5-0.20230119111129-c797534c51e4/go.mod h1:oNIeGj67SjwrRSTEukErjNT1zZ02W9+8M5mSgQCkZC8=
+github.com/vincentbernat/goflow2 v1.0.5-0.20230210215259-34e153ff515d h1:g5fI2u63E5KaS5rQlpVveKrTtKAT3XiChAeNH0TRnpk=
+github.com/vincentbernat/goflow2 v1.0.5-0.20230210215259-34e153ff515d/go.mod h1:oNIeGj67SjwrRSTEukErjNT1zZ02W9+8M5mSgQCkZC8=
 github.com/vincentbernat/patricia v0.0.0-20220923091046-b376a1167a94 h1:T7+yyM6300NYIv1kqlXX53d2cjEHpgDt6cFbBYO+upk=
 github.com/vincentbernat/patricia v0.0.0-20220923091046-b376a1167a94/go.mod h1:6jY40ESetsbfi04/S12iJlsiS6DYL2B2W+WAcqoDHtw=
 github.com/vishvananda/netns v0.0.0-20180720170159-13995c7128cc/go.mod h1:ZjcWmFBXmLKZu9Nxj3WKYEafiSqer2rnvPr0en9UNpI=

inlet/flow/decoder/sflow/root_test.go

@@ -304,4 +304,40 @@ func TestDecodeInterface(t *testing.T) {
 		}
 
 	})
+
+	t.Run("flow sample with IPv4 data", func(t *testing.T) {
+		// Send data
+		data := helpers.ReadPcapPayload(t, filepath.Join("testdata", "data-sflow-ipv4-data.pcap"))
+		got := sdecoder.Decode(decoder.RawFlow{Payload: data, Source: net.ParseIP("127.0.0.1")})
+		if got == nil {
+			t.Fatalf("Decode() error on data")
+		}
+		expectedFlows := []*schema.FlowMessage{
+			{
+				SamplingRate:    256,
+				InIf:            0,
+				OutIf:           182,
+				SrcAddr:         netip.MustParseAddr("::ffff:50.50.50.50"),
+				DstAddr:         netip.MustParseAddr("::ffff:51.51.51.51"),
+				ExporterAddress: netip.MustParseAddr("::ffff:49.49.49.49"),
+				GotASPath:       false,
+				ProtobufDebug: map[schema.ColumnKey]interface{}{
+					schema.ColumnBytes:   1390,
+					schema.ColumnPackets: 1,
+					schema.ColumnEType:   helpers.ETypeIPv4,
+					schema.ColumnProto:   17,
+					schema.ColumnSrcPort: 46622,
+					schema.ColumnDstPort: 58631,
+				},
+			},
+		}
+		for _, f := range got {
+			f.TimeReceived = 0
+		}
+
+		if diff := helpers.Diff(got, expectedFlows); diff != "" {
+			t.Fatalf("Decode() (-got, +want):\n%s", diff)
+		}
+
+	})
 }