Update parsers test to use ndjson parser

This commit updates the parse test to use ndjson parser instead of
multiline because the multiline parser can have issues when journald
input is reading from files. There is a corner case where the
journalctl exits successfully and the reader goroutine gets an error,
this makes Next to return early, making the multiline to also return
early.

So far I have only seen this happening when reading from file and at
the very end of the file, hence it does not seem to be a critical bug.

filebeat/input/journald/input_parsers_test.go

@@ -30,45 +30,42 @@ import (
 // TestInputParsers ensures journald input support parsers,
 // it only tests a single parser, but that is enough to ensure
 // we're correctly using the parsers
-//
-// TODO(Tiago): Fix "this test", well the way we read data from journalctl
-// it can happen that we get a read error when reading the stdout from journalctl
-// the error is "read |0: file already closed". It breaks this parsers/multiline
-// test because it will cause Next() to return an error, making the multiline return
-// earlier. All the messages end up being correctly read by Journald input's reader,
-// however the line aggregation is not correct.
-//
-// It's also interesting that it only seems to happen if more than one test is run
-// (like when running `go test ./...` or this test is run multiple times, by
-// passing -count to `go test`.
-// Running go run golang.org/x/tools/cmd/stress@latest ./filebeat.test -test.v -test.run=TestInputParsers
-// never causes a failure.
 func TestInputParsers(t *testing.T) {
- inputParsersExpected := []string{"1st line\n2nd line\n3rd line", "4th line\n5th line\n6th line"}
  env := newInputTestingEnvironment(t)
-
  inp := env.mustCreateInput(mapstr.M{
- "paths": []string{path.Join("testdata", "input-multiline-parser.journal")},
- "include_matches.match": []string{"_SYSTEMD_USER_UNIT=log-service.service"},
+ "paths": []string{path.Join("testdata", "ndjson-parser.journal")},
  "parsers": []mapstr.M{
  {
- "multiline": mapstr.M{
- "type": "count",
- "count_lines": 3,
+ "ndjson": mapstr.M{
+ "target": "",
  },
  },
  },
  })
 
  ctx, cancelInput := context.WithCancel(context.Background())
+ t.Cleanup(cancelInput)
  env.startInput(ctx, inp)
- env.waitUntilEventCount(len(inputParsersExpected))
+ env.waitUntilEventCount(1)
+ event := env.pipeline.clients[0].GetEvents()[0]
 
- for idx, event := range env.pipeline.clients[0].GetEvents() {
- if got, expected := event.Fields["message"], inputParsersExpected[idx]; got != expected {
- t.Errorf("expecting event message %q, got %q", expected, got)
- }
+ foo, isString := event.Fields["foo"].(string)
+ if !isString {
+ t.Errorf("expecting field 'foo' to be string, got %T", event.Fields["foo"])
  }
 
- cancelInput()
+ answer, isInt := event.Fields["answer"].(int64)
+ if !isInt {
+ t.Errorf("expecting field 'answer' to be int64, got %T", event.Fields["answer"])
+ }
+
+ // The JSON in the test journal is: '{"foo": "bar", "answer":42}'
+ expectedFoo := "bar"
+ expectedAnswer := int64(42)
+ if foo != expectedFoo {
+ t.Errorf("expecting foo to be '%s' got '%s' instead", expectedFoo, foo)
+ }
+ if answer != expectedAnswer {
+ t.Errorf("expecting foo to be '%d' got '%d' instead", expectedAnswer, answer)
+ }
 }