check stream name in cri decoder (#633)

* check stream name in cri decoder

* try to split in cri decode

decoder/cri.go

@@ -28,14 +28,18 @@ func DecodeCRI(data []byte) (row CRIRow, _ error) {
 	row.Time = data[:pos]
 	data = data[pos+1:]
 
-	// stream type
-	pos = bytes.IndexByte(data, criDelimiter)
-	if pos < 0 {
-		return row, fmt.Errorf("stream type is not found")
+	var stream []byte
+	// stderr or stdout
+	for len(stream) != 6 {
+		// stream type
+		pos = bytes.IndexByte(data, criDelimiter)
+		if pos < 0 {
+			return row, fmt.Errorf("stream type is not found")
+		}
+		stream = data[:pos]
+		data = data[pos+1:]
 	}
-
-	row.Stream = data[:pos]
-	data = data[pos+1:]
+	row.Stream = stream
 
 	// tags
 	pos = bytes.IndexByte(data, criDelimiter)

decoder/cri_test.go

@@ -31,3 +31,29 @@ func TestCRIError(t *testing.T) {
 
 	assert.Error(t, err, "there must be an error")
 }
+
+func TestCRIErrorJoined2Lines(t *testing.T) {
+	_, err := DecodeCRI([]byte("2016-10-06T00:17:09.669794202Z s2024-05-22T10:15:04.129321194Z 3\n"))
+
+	assert.Error(t, err, "there must be an error")
+}
+
+func TestCRIJoined2Lines(t *testing.T) {
+	row, err := DecodeCRI([]byte("2024-05-22T09:51:04.025764351Z s2024-05-22T10:15:04.129321194Z stderr F 2024/05/22 10:15:04 start prepraring file\n"))
+
+	assert.NoError(t, err, "error while decoding cri log")
+	assert.Equal(t, "2024-05-22T09:51:04.025764351Z", string(row.Time))
+	assert.Equal(t, "stderr", string(row.Stream))
+	assert.Equal(t, "2024/05/22 10:15:04 start prepraring file\n", string(row.Log))
+	assert.Equal(t, false, row.IsPartial)
+}
+
+func TestCRIJoined3Lines(t *testing.T) {
+	row, err := DecodeCRI([]byte(`2024-05-22T09:51:04.025764351Z s2024-05-22T10:15:04.129321194Z stder2024-05-22T01:16:15.512917014Z stdo 2024-05-22T06:39:29.230708414Z stdout F {"level":"warn","ts":"2024-05-22T06:39:29.230Z"}\n`))
+
+	assert.NoError(t, err, "error while decoding cri log")
+	assert.Equal(t, "2024-05-22T09:51:04.025764351Z", string(row.Time))
+	assert.Equal(t, "stdout", string(row.Stream))
+	assert.Equal(t, `{"level":"warn","ts":"2024-05-22T06:39:29.230Z"}\n`, string(row.Log))
+	assert.Equal(t, false, row.IsPartial)
+}