Fix panic when parsing invalid lines

The line

```
|h|#consumer:Kafka::SharedConfigurationConsumer,topic:shared_configuration_update,partition:1,consumer_group:tc_rc_us
```

caused a panic because the line parsing _first_ splits by `:` and then failed to
find a `|` to split on.

Check that we get at least two "line parts" (i.e. splits around `|`) when we
expect them, and if not, gracefully reject the line instead of crashing.

Fixes #572.

Signed-off-by: Matthias Rampke <matthias@prometheus.io>

pkg/line/line.go

@@ -229,6 +229,11 @@ func (p *Parser) LineToEvents(line string, sampleErrors prometheus.CounterVec, s
 
 	var samples []string
 	lineParts := strings.SplitN(elements[1], "|", 3)
+	if len(lineParts) < 2 {
+		sampleErrors.WithLabelValues("not_enough_parts_after_colon").Inc()
+		level.Debug(logger).Log("msg", "Bad line (not enough parts after first :) from StatsD", "line", line)
+		return events
+	}
 	if strings.Contains(lineParts[0], ":") {
 		// handle DogStatsD extended aggregation
 		isValidAggType := false

pkg/line/line_test.go

@@ -823,6 +823,12 @@ func TestLineToEvents(t *testing.T) {
 				},
 			},
 		},
+		"invalid event split over lines part 1": {
+			in: "karafka.consumer.consume.cpu_idle_second:  0.111090  -0.055903  -0.195390 (  2.419002)",
+		},
+		"invalid event split over lines part 2": {
+			in: "|h|#consumer:Kafka::SharedConfigurationConsumer,topic:shared_configuration_update,partition:1,consumer_group:tc_rc_us",
+		},
 	}
 
 	parser := NewParser()