[receiver/filelog] Add ability to set log body when parsing (open-tel…

…emetry#13130)

This follows roughly the same pattern as other "sub-parsers".
The idea here is that a parser isolates individual values, which
may then be assigned to specific fields in the data model. Although
there are other ways, users often wish to perform these assignments
immediately after isolating the values, as it is a logical next
step and more concise.

pkg/stanza/docs/types/parsers.md

@@ -53,3 +53,4 @@ List of embeddable operations:
 - [`severity`](./severity.md)
 - [`trace`](./trace.md)
 - [`scope_name`](./scope_name.md)
+- `body`: A field that should be assigned to a the log body.

pkg/stanza/operator/helper/parser.go

@@ -16,6 +16,7 @@ package helper // import "github.com/open-telemetry/opentelemetry-collector-cont
 
 import (
  "context"
+ "fmt"
 
  "go.uber.org/zap"
 
@@ -38,6 +39,7 @@ type ParserConfig struct {
 
  ParseFrom entry.Field `mapstructure:"parse_from" json:"parse_from" yaml:"parse_from"`
  ParseTo entry.Field `mapstructure:"parse_to" json:"parse_to" yaml:"parse_to"`
+ BodyField *entry.Field `mapstructure:"body" json:"body" yaml:"body"`
  TimeParser *TimeParser `mapstructure:"timestamp,omitempty" json:"timestamp,omitempty" yaml:"timestamp,omitempty"`
  Config *SeverityConfig `mapstructure:"severity,omitempty" json:"severity,omitempty" yaml:"severity,omitempty"`
  TraceParser *TraceParser `mapstructure:"trace,omitempty" json:"trace,omitempty" yaml:"trace,omitempty"`
@@ -51,10 +53,15 @@ func (c ParserConfig) Build(logger *zap.SugaredLogger) (ParserOperator, error) {
  return ParserOperator{}, err
  }
 
+ if c.BodyField != nil && c.ParseTo.String() == entry.NewBodyField().String() {
+ return ParserOperator{}, fmt.Errorf("`parse_to: body` not allowed when `body` is configured")
+ }
+
  parserOperator := ParserOperator{
  TransformerOperator: transformerOperator,
  ParseFrom: c.ParseFrom,
  ParseTo: c.ParseTo,
+ BodyField: c.BodyField,
  }
 
  if c.TimeParser != nil {
@@ -91,6 +98,7 @@ type ParserOperator struct {
  TransformerOperator
  ParseFrom entry.Field
  ParseTo entry.Field
+ BodyField *entry.Field
  TimeParser *TimeParser
  SeverityParser *SeverityParser
  TraceParser *TraceParser
@@ -148,6 +156,12 @@ func (p *ParserOperator) ParseWith(ctx context.Context, entry *entry.Entry, pars
  return p.HandleEntryError(ctx, entry, errors.Wrap(err, "set parse_to"))
  }
 
+ if p.BodyField != nil {
+ if body, ok := p.BodyField.Get(entry); ok {
+ entry.Body = body
+ }
+ }
+
  var timeParseErr error
  if p.TimeParser != nil {
  timeParseErr = p.TimeParser.Parse(entry)
@@ -168,7 +182,7 @@ func (p *ParserOperator) ParseWith(ctx context.Context, entry *entry.Entry, pars
  scopeNameParserErr = p.ScopeNameParser.Parse(entry)
  }
 
- // Handle time or severity parsing errors after attempting to parse both
+ // Handle parsing errors after attempting to parse all
  if timeParseErr != nil {
  return p.HandleEntryError(ctx, entry, errors.Wrap(timeParseErr, "time parser"))
  }

pkg/stanza/operator/helper/parser_test.go

@@ -52,6 +52,18 @@ func TestParserConfigInvalidTimeParser(t *testing.T) {
  require.Contains(t, err.Error(), "missing required configuration parameter `layout`")
 }
 
+func TestParserConfigBodyCollision(t *testing.T) {
+ cfg := NewParserConfig("test-id", "test-type")
+ cfg.ParseTo = entry.NewBodyField()
+
+ b := entry.NewAttributeField("message")
+ cfg.BodyField = &b
+
+ _, err := cfg.Build(testutil.Logger(t))
+ require.Error(t, err)
+ require.Contains(t, err.Error(), "`parse_to: body` not allowed when `body` is configured")
+}
+
 func TestParserConfigBuildValid(t *testing.T) {
  cfg := NewParserConfig("test-id", "test-type")
 
@@ -508,6 +520,28 @@ func TestParserFields(t *testing.T) {
  return e
  },
  },
+ {
+ "ParseAndSetBody",
+ func(cfg *ParserConfig) {
+ b := entry.NewAttributeField("key")
+ cfg.BodyField = &b
+ },
+ func() *entry.Entry {
+ e := entry.New()
+ e.ObservedTimestamp = now
+ e.Body = keyValue
+ return e
+ },
+ func() *entry.Entry {
+ e := entry.New()
+ e.ObservedTimestamp = now
+ e.Attributes = map[string]interface{}{
+ "key": "value",
+ }
+ e.Body = "value"
+ return e
+ },
+ },
  {
  "ParseFromBodyField",
  func(cfg *ParserConfig) {

unreleased/pkg-stanza-body-block.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: "`filelog`, `journald`, `syslog`, `tcplog`, `udplog`, `windowseventlog` receivers"
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add ability to set log body when when parsing.
+
+# One or more tracking issues related to the change
+issues: [10274]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: