Move parsers outside of filestream input so others can use them as we…

…ll (#26541)

## What does this PR do?

The object has its own `Unpack` function, so it is enough for you to add it as an attribute to your configuration.

```golang
config parser.Config
```

Then create the parser based on the configuration

```golang
p = inp.parserConfig.Create(r)
```

Example configuration accepted by the code above

```yaml
parsers:
  - multiline:
      type: count
      lines_count: 3
```

(cherry picked from commit 2a56cd7)

filebeat/input/filestream/config.go

@@ -25,6 +25,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"
 )
 
@@ -71,7 +72,7 @@ type readerConfig struct {
  MaxBytes int `config:"message_max_bytes" validate:"min=0,nonzero"`
  Tail bool `config:"seek_to_tail"`
 
- Parsers []common.ConfigNamespace `config:"parsers"`
+ Parsers parser.Config `config:",inline"`
 }
 
 type backoffConfig struct {
@@ -127,7 +128,6 @@ func defaultReaderConfig() readerConfig {
  LineTerminator: readfile.AutoLineTerminator,
  MaxBytes: 10 * humanize.MiByte,
  Tail: false,
- Parsers: make([]common.ConfigNamespace, 0),
  }
 }
 
@@ -136,9 +136,5 @@ func (c *config) Validate() error {
  return fmt.Errorf("no path is configured")
  }
 
- if err := validateParserConfig(parserConfig{maxBytes: c.Reader.MaxBytes, lineTerminator: c.Reader.LineTerminator}, c.Reader.Parsers); err != nil {
- return fmt.Errorf("cannot parse parser configuration: %+v", err)
- }
-
  return nil
 }

filebeat/input/filestream/input.go

@@ -35,6 +35,7 @@ import (
  "github.com/elastic/beats/v7/libbeat/logp"
  "github.com/elastic/beats/v7/libbeat/reader"
  "github.com/elastic/beats/v7/libbeat/reader/debug"
+ "github.com/elastic/beats/v7/libbeat/reader/parser"
  "github.com/elastic/beats/v7/libbeat/reader/readfile"
  "github.com/elastic/beats/v7/libbeat/reader/readfile/encoding"
 )
@@ -57,7 +58,7 @@ type filestream struct {
  encodingFactory encoding.EncodingFactory
  encoding encoding.Encoding
  closerConfig closerConfig
- parserConfig []common.ConfigNamespace
+ parsers  parser.Config
 }
 
 // Plugin creates a new filestream input plugin for creating a stateful input.
@@ -97,6 +98,7 @@ func configure(cfg *common.Config) (loginp.Prospector, loginp.Harvester, error)
  readerConfig: config.Reader,
  encodingFactory: encodingFactory,
  closerConfig: config.Close,
+ parsers: config.Reader.Parsers,
  }
 
  return prospector, filestream, nil
@@ -219,10 +221,7 @@ func (inp *filestream) open(log *logp.Logger, canceler input.Canceler, fs fileSo
 
  r = readfile.NewFilemeta(r, fs.newPath)
 
- r, err = newParsers(r, parserConfig{maxBytes: inp.readerConfig.MaxBytes, lineTerminator: inp.readerConfig.LineTerminator}, inp.readerConfig.Parsers)
- if err != nil {
- return nil, err
- }
+ r = inp.parsers.Create(r)
 
  r = readfile.NewLimitReader(r, inp.readerConfig.MaxBytes)
 

filebeat/input/filestream/parser.go → libbeat/reader/parser/parser.go

@@ -15,13 +15,15 @@
 // specific language governing permissions and limitations
 // under the License.
 
-package filestream
+package parser
 
 import (
  "errors"
  "fmt"
  "io"
 
+ "github.com/dustin/go-humanize"
+
  "github.com/elastic/beats/v7/libbeat/common"
  "github.com/elastic/beats/v7/libbeat/reader"
  "github.com/elastic/beats/v7/libbeat/reader/multiline"
@@ -35,20 +37,48 @@ var (
 
 // parser transforms or translates the Content attribute of a Message.
 // They are able to aggregate two or more Messages into a single one.
-type parser interface {
+type Parser interface {
  io.Closer
  Next() (reader.Message, error)
 }
 
-type parserConfig struct {
- maxBytes int
- lineTerminator readfile.LineTerminator
+type CommonConfig struct {
+ MaxBytes int `config:"max_bytes"`
+ LineTerminator readfile.LineTerminator `config:"line_terminator"`
 }
 
-func newParsers(in reader.Reader, pCfg parserConfig, c []common.ConfigNamespace) (parser, error) {
- p := in
+type Config struct {
+ pCfg CommonConfig
+ parsers []common.ConfigNamespace
+}
+
+func (c *Config) Unpack(cc *common.Config) error {
+ tmp := struct {
+ Common CommonConfig `config:",inline"`
+ Parsers []common.ConfigNamespace `config:"parsers"`
+ }{
+ CommonConfig{
+ MaxBytes: 10 * humanize.MiByte,
+ LineTerminator: readfile.AutoLineTerminator,
+ },
+ nil,
+ }
+ err := cc.Unpack(&tmp)
+ if err != nil {
+ return err
+ }
 
- for _, ns := range c {
+ newC, err := NewConfig(tmp.Common, tmp.Parsers)
+ if err != nil {
+ return err
+ }
+ *c = *newC
+
+ return nil
+}
+
+func NewConfig(pCfg CommonConfig, parsers []common.ConfigNamespace) (*Config, error) {
+ for _, ns := range parsers {
  name := ns.Name()
  switch name {
  case "multiline":
@@ -58,63 +88,68 @@ func newParsers(in reader.Reader, pCfg parserConfig, c []common.ConfigNamespace)
  if err != nil {
  return nil, fmt.Errorf("error while parsing multiline parser config: %+v", err)
  }
- p, err = multiline.New(p, "\n", pCfg.maxBytes, &config)
- if err != nil {
- return nil, fmt.Errorf("error while creating multiline parser: %+v", err)
- }
  case "ndjson":
  var config readjson.ParserConfig
  cfg := ns.Config()
  err := cfg.Unpack(&config)
  if err != nil {
  return nil, fmt.Errorf("error while parsing ndjson parser config: %+v", err)
  }
- p = readjson.NewJSONParser(p, &config)
  case "container":
  config := readjson.DefaultContainerConfig()
  cfg := ns.Config()
  err := cfg.Unpack(&config)
  if err != nil {
  return nil, fmt.Errorf("error while parsing container parser config: %+v", err)
  }
- p = readjson.NewContainerParser(p, &config)
  default:
  return nil, fmt.Errorf("%s: %s", ErrNoSuchParser, name)
  }
  }
 
- return p, nil
+ return &Config{
+ pCfg: pCfg,
+ parsers: parsers,
+ }, nil
+
 }
 
-func validateParserConfig(pCfg parserConfig, c []common.ConfigNamespace) error {
- for _, ns := range c {
+func (c *Config) Create(in reader.Reader) Parser {
+ p := in
+ for _, ns := range c.parsers {
  name := ns.Name()
  switch name {
  case "multiline":
  var config multiline.Config
  cfg := ns.Config()
  err := cfg.Unpack(&config)
  if err != nil {
- return fmt.Errorf("error while parsing multiline parser config: %+v", err)
+ return p
+ }
+ p, err = multiline.New(p, "\n", c.pCfg.MaxBytes, &config)
+ if err != nil {
+ return p
  }
  case "ndjson":
- var config readjson.Config
+ var config readjson.ParserConfig
  cfg := ns.Config()
  err := cfg.Unpack(&config)
  if err != nil {
- return fmt.Errorf("error while parsing ndjson parser config: %+v", err)
+ return p
  }
+ p = readjson.NewJSONParser(p, &config)
  case "container":
  config := readjson.DefaultContainerConfig()
  cfg := ns.Config()
  err := cfg.Unpack(&config)
  if err != nil {
- return fmt.Errorf("error while parsing container parser config: %+v", err)
+ return p
  }
+ p = readjson.NewContainerParser(p, &config)
  default:
- return fmt.Errorf("%s: %s", ErrNoSuchParser, name)
+ return p
  }
  }
 
- return nil
+ return p
 }

libbeat/reader/parser/parser_example_test.go

@@ -0,0 +1,94 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package parser
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/elastic/beats/v7/libbeat/common"
+ "github.com/elastic/beats/v7/libbeat/reader/readfile"
+)
+
+type inputParsersConfig struct {
+ MaxBytes int `config:"max_bytes"`
+ LineTerminator readfile.LineTerminator `config:"line_terminator"`
+ Parsers Config `config:",inline"`
+}
+
+func TestParsersExampleInline(t *testing.T) {
+ tests := map[string]struct {
+ lines string
+ parsers map[string]interface{}
+ expectedMessages []string
+ }{
+ "multiline docker logs parser": {
+ lines: `{"log":"[log] The following are log messages\n","stream":"stdout","time":"2016-03-02T22:58:51.338462311Z"}
+{"log":"[log] This one is\n","stream":"stdout","time":"2016-03-02T22:58:51.338462311Z"}
+{"log":" on multiple\n","stream":"stdout","time":"2016-03-02T22:58:51.338462311Z"}
+{"log":" lines","stream":"stdout","time":"2016-03-02T22:58:51.338462311Z"}
+{"log":"[log] In total there should be 3 events\n","stream":"stdout","time":"2016-03-02T22:58:51.338462311Z"}
+`,
+ parsers: map[string]interface{}{
+ "max_bytes": 1024,
+ "line_terminator": "auto",
+ "parsers": []map[string]interface{}{
+ map[string]interface{}{
+ "ndjson": map[string]interface{}{
+ "keys_under_root": true,
+ "message_key": "log",
+ },
+ },
+ map[string]interface{}{
+ "multiline": map[string]interface{}{
+ "match": "after",
+ "negate": true,
+ "pattern": "^\\[log\\]",
+ },
+ },
+ },
+ },
+ expectedMessages: []string{
+ "[log] The following are log messages\n",
+ "[log] This one is\n\n on multiple\n\n lines",
+ "[log] In total there should be 3 events\n",
+ },
+ },
+ }
+
+ for name, test := range tests {
+ test := test
+ t.Run(name, func(t *testing.T) {
+ cfg := common.MustNewConfigFrom(test.parsers)
+ var c inputParsersConfig
+ err := cfg.Unpack(&c)
+ require.NoError(t, err)
+
+ p := c.Parsers.Create(testReader(test.lines))
+
+ i := 0
+ msg, err := p.Next()
+ for err == nil {
+ require.Equal(t, test.expectedMessages[i], string(msg.Content))
+ i++
+ msg, err = p.Next()
+ }
+ })
+ }
+}