x-pack/filebeat/input/httpjson/internal/v2: allow split chains to con…

…tinue past empty targets (#27880) (#27914)

This adds a configuration option "ignore_empty_value" that allows a split
processor chain to continue if a target field is missing or empty.

Updates #26008

(cherry picked from commit 2036ad8)

Co-authored-by: Dan Kortschak <90160302+efd6@users.noreply.github.com>
Co-authored-by: Dan Kortschak <dan.kortschak@elastic.co>

CHANGELOG.next.asciidoc

@@ -433,6 +433,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add base64 Encode functionality to httpjson input. {pull}27681[27681]
 - Add `join` and `sprintf` functions to `httpjson` input. {pull}27735[27735]
 - Improve memory usage of line reader of `log` and `filestream` input. {pull}27782[27782]
+- Add `ignore_empty_value` flag to `httpjson` `split` processor. {pull}27880[27880]
 
 - Add `ignore_empty_value` flag to `httpjson` `split` processor. {pull}27880[27880]
 - Update Cisco ASA/FTD ingest pipeline grok/dissect patterns for multiple message IDs. {issue}26869[26869] {pull}26879[26879]

x-pack/filebeat/docs/inputs/input-httpjson.asciidoc

@@ -566,6 +566,11 @@ Required if using split type of `string`. This is the sub string used to split
 
 Valid when used with `type: map`. When not empty, defines a new field where the original key value will be stored.
 
+[float]
+==== `response.split[].ignore_empty_value`
+
+If set to true, empty or missing value will be ignored and processing will pass on to the next nested split operation instead of failing with an error. Default: `false`.
+
 [float]
 ==== `response.split[].split`
 

x-pack/filebeat/input/httpjson/internal/v2/config_response.go

@@ -24,13 +24,14 @@ type responseConfig struct {
 }
 
 type splitConfig struct {
- Target string `config:"target" validation:"required"`
- Type string `config:"type"`
- Transforms transformsConfig `config:"transforms"`
- Split *splitConfig `config:"split"`
- KeepParent bool `config:"keep_parent"`
- KeyField string `config:"key_field"`
- DelimiterString string `config:"delimiter"`
+ Target string `config:"target" validation:"required"`
+ Type string `config:"type"`
+ Transforms transformsConfig `config:"transforms"`
+ Split *splitConfig `config:"split"`
+ KeepParent bool `config:"keep_parent"`
+ KeyField string `config:"key_field"`
+ DelimiterString string `config:"delimiter"`
+ IgnoreEmptyValue bool `config:"ignore_empty_value"`
 }
 
 func (c *responseConfig) Validate() error {

x-pack/filebeat/input/httpjson/internal/v2/split.go

@@ -21,18 +21,24 @@ var (
  errExpectedSplitString = errors.New("split was expecting field to be a string")
 )
 
+// split is a split processor chain element. Split processing is executed
+// by applying elements of the chain's linked list to an input until completed
+// or an error state is encountered.
 type split struct {
- log *logp.Logger
- targetInfo targetInfo
- kind string
- transforms []basicTransform
- child *split
- keepParent bool
- keyField string
- isRoot bool
- delimiter string
+ log *logp.Logger
+ targetInfo targetInfo
+ kind string
+ transforms []basicTransform
+ child *split
+ keepParent bool
+ ignoreEmptyValue bool
+ keyField string
+ isRoot bool
+ delimiter string
 }
 
+// newSplitResponse returns a new split based on the provided config and
+// logging to the provided logger, tagging the split as the root of the chain.
 func newSplitResponse(cfg *splitConfig, log *logp.Logger) (*split, error) {
  if cfg == nil {
  return nil, nil
@@ -42,11 +48,13 @@ func newSplitResponse(cfg *splitConfig, log *logp.Logger) (*split, error) {
  if err != nil {
  return nil, err
  }
- // we want to be able to identify which split is the root of the chain
+ // We want to be able to identify which split is the root of the chain.
  split.isRoot = true
  return split, nil
 }
 
+// newSplit returns a new split based on the provided config and
+// logging to the provided logger.
 func newSplit(c *splitConfig, log *logp.Logger) (*split, error) {
  ti, err := getTargetInfo(c.Target)
  if err != nil {
@@ -71,29 +79,40 @@ func newSplit(c *splitConfig, log *logp.Logger) (*split, error) {
  }
 
  return &split{
- log: log,
- targetInfo: ti,
- kind: c.Type,
- keepParent: c.KeepParent,
- keyField: c.KeyField,
- delimiter: c.DelimiterString,
- transforms: ts,
- child: s,
+ log: log,
+ targetInfo: ti,
+ kind: c.Type,
+ keepParent: c.KeepParent,
+ ignoreEmptyValue: c.IgnoreEmptyValue,
+ keyField: c.KeyField,
+ delimiter: c.DelimiterString,
+ transforms: ts,
+ child: s,
  }, nil
 }
 
+// run runs the split operation on the contents of resp, sending successive
+// split results on ch. ctx is passed to transforms that are called during
+// the split.
 func (s *split) run(ctx *transformContext, resp transformable, ch chan<- maybeMsg) error {
  root := resp.body()
  return s.split(ctx, root, ch)
 }
 
+// split recursively executes the split processor chain.
 func (s *split) split(ctx *transformContext, root common.MapStr, ch chan<- maybeMsg) error {
  v, err := root.GetValue(s.targetInfo.Name)
  if err != nil && err != common.ErrKeyNotFound {
  return err
  }
 
  if v == nil {
+ if s.ignoreEmptyValue {
+ if s.child != nil {
+ return s.child.split(ctx, root, ch)
+ }
+ return nil
+ }
  if s.isRoot {
  return errEmptyRootField
  }
@@ -109,6 +128,12 @@ func (s *split) split(ctx *transformContext, root common.MapStr, ch chan<- maybe
  }
 
  if len(varr) == 0 {
+ if s.ignoreEmptyValue {
+ if s.child != nil {
+ return s.child.split(ctx, root, ch)
+ }
+ return nil
+ }
  if s.isRoot {
  return errEmptyRootField
  }
@@ -130,6 +155,12 @@ func (s *split) split(ctx *transformContext, root common.MapStr, ch chan<- maybe
  }
 
  if len(vmap) == 0 {
+ if s.ignoreEmptyValue {
+ if s.child != nil {
+ return s.child.split(ctx, root, ch)
+ }
+ return nil
+ }
  if s.isRoot {
  return errEmptyRootField
  }
@@ -151,6 +182,12 @@ func (s *split) split(ctx *transformContext, root common.MapStr, ch chan<- maybe
  }
 
  if len(vstr) == 0 {
+ if s.ignoreEmptyValue {
+ if s.child != nil {
+ return s.child.split(ctx, root, ch)
+ }
+ return nil
+ }
  if s.isRoot {
  return errEmptyRootField
  }
@@ -169,6 +206,8 @@ func (s *split) split(ctx *transformContext, root common.MapStr, ch chan<- maybe
  return errors.New("unknown split type")
 }
 
+// sendMessage sends an array or map split result value, v, on ch after performing
+// any necessary transformations. If key is "", the value is an element of an array.
 func (s *split) sendMessage(ctx *transformContext, root common.MapStr, key string, v interface{}, ch chan<- maybeMsg) error {
  obj, ok := toMapStr(v)
  if !ok {
@@ -220,6 +259,8 @@ func toMapStr(v interface{}) (common.MapStr, bool) {
  return common.MapStr{}, false
 }
 
+// sendMessage sends a string split result value, v, on ch after performing any
+// necessary transformations. If key is "", the value is an element of an array.
 func (s *split) sendMessageSplitString(ctx *transformContext, root common.MapStr, v string, ch chan<- maybeMsg) error {
  clone := root.Clone()
  _, _ = clone.Put(s.targetInfo.Name, v)