Goflow fix failed to parse ip (#336)

* log correct key that failed to parse as an ip * handle ip address fields that are set but empty * 1.1.2 changelog
observIQ · Jun 24, 2021 · d0de115 · d0de115
1 parent f1682b4
commit d0de115
Show file tree

Hide file tree

Showing 3 changed files with 138 additions and 2 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -4,7 +4,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## Unreleased
+## 1.1.2 - Unreleased
+
+### Fixed
+- Resolved an issue where empty ip address fields result in failed parsing [PR 336](https://github.com/observIQ/stanza/pull/336)
 
 ## 1.1.1 - 2021-06-21
 

diff --git a/operator/builtin/input/goflow/parse.go b/operator/builtin/input/goflow/parse.go
@@ -33,9 +33,13 @@ func Parse(message flowmessage.FlowMessage) (map[string]interface{}, time.Time,
 			delete(m, key)
 			switch x := val.(type) {
 			case []byte:
+				// If the field is not set, skip it
+				if len(x) == 0 {
+					continue
+				}
 				ip, err := bytesToIP(x)
 				if err != nil {
-					return nil, timestamp, errors.Wrap(err, "error converting DstAddr to string")
+					return nil, timestamp, errors.Wrap(err, fmt.Sprintf("error converting %s to string", key))
 				}
 				m[key] = ip.String()
 			default:

diff --git a/operator/builtin/input/goflow/parse_test.go b/operator/builtin/input/goflow/parse_test.go
@@ -25,12 +25,14 @@ func TestParse(t *testing.T) {
 		input      flowmessage.FlowMessage
 		expect     map[string]interface{}
 		expectTime time.Time
+		expectErr  bool
 	}{
 		{
 			"minimal",
 			flowmessage.FlowMessage{},
 			map[string]interface{}{},
 			time.Time{},
+			false,
 		},
 		{
 			"addresses",
@@ -51,6 +53,128 @@ func TestParse(t *testing.T) {
 				"dstaddrencap":   DstAddrEncap.String(),
 			},
 			time.Time{},
+			false,
+		},
+		{
+			"empty-srcaddr",
+			flowmessage.FlowMessage{
+				SamplerAddress: SamplerAddress,
+				SrcAddr:        []byte{},
+				DstAddr:        DstAddr,
+				NextHop:        NextHop,
+				SrcAddrEncap:   SrcAddrEncap,
+				DstAddrEncap:   DstAddrEncap,
+			},
+			map[string]interface{}{
+				"sampleraddress": SamplerAddress.String(),
+				"dstaddr":        DstAddr.String(),
+				"nexthop":        NextHop.String(),
+				"srcaddrencap":   SrcAddrEncap.String(),
+				"dstaddrencap":   DstAddrEncap.String(),
+			},
+			time.Time{},
+			false,
+		},
+		{
+			"empty-dstaddr",
+			flowmessage.FlowMessage{
+				SamplerAddress: SamplerAddress,
+				SrcAddr:        SrcAddr,
+				DstAddr:        []byte{},
+				NextHop:        NextHop,
+				SrcAddrEncap:   SrcAddrEncap,
+				DstAddrEncap:   DstAddrEncap,
+			},
+			map[string]interface{}{
+				"sampleraddress": SamplerAddress.String(),
+				"srcaddr":        SrcAddr.String(),
+				"nexthop":        NextHop.String(),
+				"srcaddrencap":   SrcAddrEncap.String(),
+				"dstaddrencap":   DstAddrEncap.String(),
+			},
+			time.Time{},
+			false,
+		},
+		{
+			"empty-nexthop",
+			flowmessage.FlowMessage{
+				SamplerAddress: SamplerAddress,
+				SrcAddr:        SrcAddr,
+				DstAddr:        DstAddr,
+				NextHop:        []byte{},
+				SrcAddrEncap:   SrcAddrEncap,
+				DstAddrEncap:   DstAddrEncap,
+			},
+			map[string]interface{}{
+				"sampleraddress": SamplerAddress.String(),
+				"srcaddr":        SrcAddr.String(),
+				"dstaddr":        DstAddr.String(),
+				"srcaddrencap":   SrcAddrEncap.String(),
+				"dstaddrencap":   DstAddrEncap.String(),
+			},
+			time.Time{},
+			false,
+		},
+		{
+			"empty-srcaddrencap",
+			flowmessage.FlowMessage{
+				SamplerAddress: SamplerAddress,
+				SrcAddr:        SrcAddr,
+				DstAddr:        DstAddr,
+				NextHop:        NextHop,
+				SrcAddrEncap:   []byte{},
+				DstAddrEncap:   DstAddrEncap,
+			},
+			map[string]interface{}{
+				"sampleraddress": SamplerAddress.String(),
+				"srcaddr":        SrcAddr.String(),
+				"dstaddr":        DstAddr.String(),
+				"nexthop":        NextHop.String(),
+				"dstaddrencap":   DstAddrEncap.String(),
+			},
+			time.Time{},
+			false,
+		},
+		{
+			"empty-dstaddrencap",
+			flowmessage.FlowMessage{
+				SamplerAddress: SamplerAddress,
+				SrcAddr:        SrcAddr,
+				DstAddr:        DstAddr,
+				NextHop:        NextHop,
+				SrcAddrEncap:   SrcAddrEncap,
+				DstAddrEncap:   []byte{},
+			},
+			map[string]interface{}{
+				"sampleraddress": SamplerAddress.String(),
+				"srcaddr":        SrcAddr.String(),
+				"dstaddr":        DstAddr.String(),
+				"nexthop":        NextHop.String(),
+				"srcaddrencap":   SrcAddrEncap.String(),
+			},
+			time.Time{},
+			false,
+		},
+		{
+			"malformed-addresses",
+			flowmessage.FlowMessage{
+				SamplerAddress: SamplerAddress,
+				SrcAddr:        []byte("ip:10.1.1.1"),
+				DstAddr:        DstAddr,
+				NextHop:        NextHop,
+				SrcAddrEncap:   SrcAddrEncap,
+				DstAddrEncap:   DstAddrEncap,
+			},
+			map[string]interface{}{
+				"sampleraddress": SamplerAddress.String(),
+				"srcaddr":        "ip:10.1.1.1",
+				"dstaddr":        DstAddr.String(),
+				"nexthop":        NextHop.String(),
+				"srcaddrencap":   SrcAddrEncap.String(),
+				"dstaddrencap":   DstAddrEncap.String(),
+			},
+			time.Time{},
+			true,
 		},
 		{
 			"promote-time",
@@ -72,12 +196,17 @@ func TestParse(t *testing.T) {
 				"dstaddrencap":   DstAddrEncap.String(),
 			},
 			time.Unix(int64(1623774351), 0),
+			false,
 		},
 	}
 
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
 			output, outputTime, err := Parse(tc.input)
+			if tc.expectErr {
+				require.Error(t, err)
+				return
+			}
 			require.NoError(t, err)
 			require.Equal(t, tc.expect, output)