[Filebeat] change multiline configuration in awss3 input to parsers

CHANGELOG.next.asciidoc

@@ -812,7 +812,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Update PanOS module to parse Global Protect & User ID logs. {issue}24722[24722] {issue}24724[24724] {pull}24927[24927]
 - Add HMAC signature validation support for http_endpoint input. {pull}24918[24918]
 - Add new grok pattern for iptables module for Ubiquiti UDM {issue}25615[25615] {pull}25616[25616]
-- Add multiline support to aws-s3 input. {issue}25249[25249] {pull}25710[25710]
+- Add multiline support to aws-s3 input. {issue}25249[25249] {pull}25710[25710] {pull}25873[25873]
 - Add monitoring metrics to the `aws-s3` input. {pull}25711[25711]
 - Added `network.direction` fields to Zeek and Suricata modules using the `add_network_direction` processor {pull}24620[24620]
 - Add Content-Type override to aws-s3 input. {issue}25697[25697] {pull}25772[25772]

x-pack/filebeat/docs/inputs/input-aws-s3.asciidoc

@@ -112,10 +112,10 @@ setting.  If `file_selectors` is given, then any global
 `expand_event_list_from_field` value is ignored in favor of the ones
 specified in the `file_selectors`.  Regex syntax is the same as the Go
 language.  Files that don't match one of the regexes won't be
-processed.  <<input-aws-s3-content_type>>, <<input-aws-s3-include_s3_metadata>>,
-<<input-aws-s3-multiline>>, <<input-aws-s3-max_bytes>>,
-<<input-aws-s3-buffer_size>>, and <<input-aws-s3-encoding>> may also be set for
-each file selector.
+processed.  <<input-aws-s3-content_type>>, <<input-aws-s3-parsers>>,
+<<input-aws-s3-include_s3_metadata>>,<<input-aws-s3-max_bytes>>,
+<<input-aws-s3-buffer_size>>, and <<input-aws-s3-encoding>> may also
+be set for each file selector.
 
 ["source", "yml"]
 ----
@@ -166,15 +166,43 @@ The maximum number of messages to return. Amazon SQS never returns more messages
 than this value (however, fewer messages might be returned). Valid values: 1 to
 10. Default: 5.
 
-[id="input-{type}-multiline"]
+[id="input-{type}-parsers"]
 [float]
-==== `multiline`
+==== `parsers`
+
+beta[]
+
+This option expects a list of parsers that non-JSON logs go through.
+
+Available parsers:
+
+* `multiline`
+
+In this example, {beatname_uc} is reading multiline messages that
+consist of XML that start with the `<Event>` tag.
+
+["source","yaml",subs="attributes"]
+----
+{beatname_lc}.inputs:
+- type: {type}
+  ...
+  parsers:
+    - multiline:
+        pattern: "^<Event"
+        negate:  true
+        match:   after
+----
+
+See the available parser settings in detail below.
+
+[float]
+===== `multiline`
 
 beta[]
 
 Options that control how {beatname_uc} deals with log messages that span
-multiple lines. This only applies to non-JSON logs.  See <<multiline-examples>>
-for more information about configuring multiline options.
+multiple lines. See <<multiline-examples>> for more information about
+configuring multiline options.
 
 [float]
 ==== `queue_url`

x-pack/filebeat/input/awss3/collector.go

@@ -32,7 +32,6 @@ import (
 	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/logp"
 	"github.com/elastic/beats/v7/libbeat/reader"
-	"github.com/elastic/beats/v7/libbeat/reader/multiline"
 	"github.com/elastic/beats/v7/libbeat/reader/readfile"
 	"github.com/elastic/beats/v7/libbeat/reader/readfile/encoding"
 	"github.com/elastic/go-concert/unison"
@@ -438,12 +437,7 @@ func (c *s3Collector) createEventsFromS3Info(svc s3iface.ClientAPI, info s3Info,
 	}
 	r = readfile.NewStripNewline(r, info.LineTerminator)
 
-	if info.Multiline != nil {
-		r, err = multiline.New(r, "\n", int(info.MaxBytes), info.Multiline)
-		if err != nil {
-			return fmt.Errorf("error setting up multiline: %v", err)
-		}
-	}
+	r = info.Parsers.Create(r)
 
 	r = readfile.NewLimitReader(r, int(info.MaxBytes))
 

x-pack/filebeat/input/awss3/config.go

@@ -12,7 +12,7 @@ import (
 
 	"github.com/elastic/beats/v7/libbeat/common/cfgtype"
 	"github.com/elastic/beats/v7/libbeat/common/match"
-	"github.com/elastic/beats/v7/libbeat/reader/multiline"
+	"github.com/elastic/beats/v7/libbeat/reader/parser"
 	"github.com/elastic/beats/v7/libbeat/reader/readfile"
 	awscommon "github.com/elastic/beats/v7/x-pack/libbeat/common/aws"
 )
@@ -66,14 +66,14 @@ type fileSelectorConfig struct {
 
 // readerConfig defines the options for reading the content of an S3 object.
 type readerConfig struct {
-	ExpandEventListFromField string                  `config:"expand_event_list_from_field"`
 	BufferSize               cfgtype.ByteSize        `config:"buffer_size"`
-	MaxBytes                 cfgtype.ByteSize        `config:"max_bytes"`
-	Multiline                *multiline.Config       `config:"multiline"`
-	LineTerminator           readfile.LineTerminator `config:"line_terminator"`
-	Encoding                 string                  `config:"encoding"`
 	ContentType              string                  `config:"content_type"`
+	Encoding                 string                  `config:"encoding"`
+	ExpandEventListFromField string                  `config:"expand_event_list_from_field"`
 	IncludeS3Metadata        []string                `config:"include_s3_metadata"`
+	LineTerminator           readfile.LineTerminator `config:"line_terminator"`
+	MaxBytes                 cfgtype.ByteSize        `config:"max_bytes"`
+	Parsers                  parser.Config           `config:",inline"`
 }
 
 func (f *readerConfig) Validate() error {

x-pack/filebeat/input/awss3/config_test.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/common/match"
+	"github.com/elastic/beats/v7/libbeat/reader/parser"
 	"github.com/elastic/beats/v7/libbeat/reader/readfile"
 )
 
@@ -21,6 +22,10 @@ func TestConfig(t *testing.T) {
 	makeConfig := func() config {
 		// Have a separate copy of defaults in the test to make it clear when
 		// anyone changes the defaults.
+		cfg := common.MustNewConfigFrom("")
+		c := parser.Config{}
+		err := c.Unpack(cfg)
+		assert.Nil(t, err)
 		return config{
 			QueueURL:            queueURL,
 			APITimeout:          120 * time.Second,
@@ -31,6 +36,7 @@ func TestConfig(t *testing.T) {
 				BufferSize:     16 * humanize.KiByte,
 				MaxBytes:       10 * humanize.MiByte,
 				LineTerminator: readfile.AutoLineTerminator,
+				Parsers:        c,
 			},
 		}
 	}

x-pack/filebeat/input/awss3/s3_integration_test.go

@@ -86,10 +86,14 @@ func defaultTestConfig() *common.Config {
 			{
 				"regex":     strings.Replace(fileName2, ".", "\\.", -1),
 				"max_bytes": 4096,
-				"multiline": common.MapStr{
-					"pattern": "^<Event",
-					"negate":  true,
-					"match":   "after",
+				"parsers": []common.MapStr{
+					{
+						"multiline": common.MapStr{
+							"pattern": "^<Event",
+							"negate":  true,
+							"match":   "after",
+						},
+					},
 				},
 			},
 		},