feat: add skipRowOnError handling for raw line protocol files (#419)

* feat: add skipRowOnError for raw line protocol files

* use common code from influxdb instead of copying

* add test

* remove dead code comment

pkg/csv2lp/lp_reader.go

@@ -0,0 +1,44 @@
+package csv2lp
+
+import (
+	"io"
+	"log"
+
+	"github.com/influxdata/influxdb/v2/models"
+)
+
+// LineProtocolFilterReader wraps a line reader and parses points, skipping if invalid
+type LineProtocolFilterReader struct {
+	// lineReader is used to report line number of the last read CSV line
+	lineReader *LineReader
+	// LineNumber represents line number of csv.Reader, 1 is the first
+	LineNumber int
+}
+
+// LineProtocolFilter creates a reader wrapper that parses points, skipping if invalid
+func LineProtocolFilter(reader io.Reader) *LineProtocolFilterReader {
+	lineReader := NewLineReader(reader)
+	lineReader.LineNumber = 1 // start counting from 1
+	return &LineProtocolFilterReader{
+		lineReader: lineReader,
+	}
+}
+
+func (state *LineProtocolFilterReader) Read(b []byte) (int, error) {
+	for {
+		bytesRead, err := state.lineReader.Read(b)
+		if err != nil {
+			return bytesRead, err
+		}
+		state.LineNumber = state.lineReader.LastLineNumber
+		buf := b[0:bytesRead]
+		pts, err := models.ParsePoints(buf) // any time precision because we won't actually use this point
+		if err != nil {
+			log.Printf("invalid point on line %d: %v\n", state.LineNumber, err)
+			continue
+		} else if len(pts) == 0 { // no points on this line
+			continue
+		}
+		return bytesRead, nil
+	}
+}

pkg/csv2lp/lp_reader_test.go

@@ -0,0 +1,68 @@
+package csv2lp
+
+import (
+	"io"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestLineProtocolFilter(t *testing.T) {
+	var tests = []struct {
+		input    string
+		expected string
+	}{
+		{
+			strings.Join([]string{
+				"weather,location=us-midwest temperature=42 1465839830100400200",
+				"awefw.,weather,location=us-east temperature=36 1465839830100400203",
+				"weather,location=us-blah temperature=32 1465839830100400204",
+				"weather,location=us-central temperature=31 1465839830100400205",
+			}, "\n"),
+			strings.Join([]string{
+				"weather,location=us-midwest temperature=42 1465839830100400200",
+				"weather,location=us-blah temperature=32 1465839830100400204",
+				"weather,location=us-central temperature=31 1465839830100400205",
+			}, "\n"),
+		},
+		{
+			strings.Join([]string{
+				"weather,location=us-midwest temperature=42 1465839830100400200",
+				"weather,location=us-east temperature=36 1465839830100400203",
+				"weather,location=us-blah temperature=32=33 1465839830100400204",
+				"weather,,location=us-blah temperature=32 1465839830100400204",
+				"weather,location=us-central temperature=31 1465839830100400205",
+			}, "\n"),
+			strings.Join([]string{
+				"weather,location=us-midwest temperature=42 1465839830100400200",
+				"weather,location=us-east temperature=36 1465839830100400203",
+				"weather,location=us-central temperature=31 1465839830100400205",
+			}, "\n"),
+		},
+		{
+			strings.Join([]string{
+				"weather,location=us-midwest temperature=42 1465839830100400200",
+				"awefw.,weather,location=us-east temperature=36 1465839830100400203",
+				"    weather,location=us-blah temperature=32 1465839830100400204",
+				"weather,location=us-east temperature=36 1465839830100400203 13413413",
+				"weather,location=us-central temperature=31 1465839830100400205",
+				"# this is a comment",
+			}, "\n"),
+			strings.Join([]string{
+				"weather,location=us-midwest temperature=42 1465839830100400200",
+				"    weather,location=us-blah temperature=32 1465839830100400204",
+				"weather,location=us-central temperature=31 1465839830100400205",
+			}, "\n"),
+		},
+	}
+	for _, tt := range tests {
+		reader := LineProtocolFilter(strings.NewReader(tt.input))
+		b, err := io.ReadAll(reader)
+		if err != nil {
+			t.Errorf("failed reading: %v", err)
+			continue
+		}
+		require.Equal(t, strings.TrimSpace(string(b)), strings.TrimSpace(tt.expected))
+	}
+}