-
Notifications
You must be signed in to change notification settings - Fork 4.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
A new processor is introduced as part of support for keeping orignal messages. Options and naming follows the convention of other processors.
### `copy_fields`
This processor copies one field to another. Example configuration is below:
```yaml
processors:
- copy_fields:
fields:
- from: message
to: event.original
fail_on_error: false
ignore_missing: true
```- Loading branch information
Showing
17 changed files
with
385 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,108 @@ | ||
| // 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 actions | ||
|
|
||
| import ( | ||
| "fmt" | ||
|
|
||
| "github.com/pkg/errors" | ||
|
|
||
| "github.com/elastic/beats/libbeat/beat" | ||
| "github.com/elastic/beats/libbeat/common" | ||
| "github.com/elastic/beats/libbeat/logp" | ||
| "github.com/elastic/beats/libbeat/processors" | ||
| ) | ||
|
|
||
| type copyFields struct { | ||
| config copyFieldsConfig | ||
| } | ||
|
|
||
| type copyFieldsConfig struct { | ||
| Fields []fromTo `config:"fields"` | ||
| IgnoreMissing bool `config:"ignore_missing"` | ||
| FailOnError bool `config:"fail_on_error"` | ||
| } | ||
|
|
||
| func init() { | ||
| processors.RegisterPlugin("copy_fields", | ||
| configChecked(newCopyFields, | ||
| requireFields("fields"), | ||
| ), | ||
| ) | ||
| } | ||
|
|
||
| func newCopyFields(c *common.Config) (processors.Processor, error) { | ||
| config := copyFieldsConfig{ | ||
| IgnoreMissing: false, | ||
| FailOnError: true, | ||
| } | ||
| err := c.Unpack(&config) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to unpack the configuration of copy processor: %s", err) | ||
| } | ||
|
|
||
| f := ©Fields{ | ||
| config: config, | ||
| } | ||
| return f, nil | ||
| } | ||
|
|
||
| func (f *copyFields) Run(event *beat.Event) (*beat.Event, error) { | ||
| var backup common.MapStr | ||
| if f.config.FailOnError { | ||
| backup = event.Fields.Clone() | ||
| } | ||
|
|
||
| for _, field := range f.config.Fields { | ||
| err := f.copyField(field.From, field.To, event.Fields) | ||
| if err != nil && f.config.FailOnError { | ||
| errMsg := fmt.Errorf("Failed to copy fields in copy_fields processor: %s", err) | ||
| logp.Debug("copy_fields", errMsg.Error()) | ||
| event.Fields = backup | ||
| event.PutValue("error.message", errMsg.Error()) | ||
| return event, err | ||
| } | ||
| } | ||
|
|
||
| return event, nil | ||
| } | ||
|
|
||
| func (f *copyFields) copyField(from string, to string, fields common.MapStr) error { | ||
| exists, _ := fields.HasKey(to) | ||
| if exists { | ||
| return fmt.Errorf("target field %s already exists, drop or rename this field first", to) | ||
| } | ||
|
|
||
| value, err := fields.GetValue(from) | ||
| if err != nil { | ||
| if f.config.IgnoreMissing && errors.Cause(err) == common.ErrKeyNotFound { | ||
| return nil | ||
| } | ||
| return fmt.Errorf("could not fetch value for key: %s, Error: %s", from, err) | ||
| } | ||
|
|
||
| _, err = fields.Put(to, value) | ||
| if err != nil { | ||
| return fmt.Errorf("could not copy value to %s: %v, %+v", to, value, err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (f *copyFields) String() string { | ||
| return "copy_fields=" + fmt.Sprintf("%+v", f.config.Fields) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,145 @@ | ||
| // 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 actions | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
|
|
||
| "github.com/elastic/beats/libbeat/beat" | ||
| "github.com/elastic/beats/libbeat/common" | ||
| ) | ||
|
|
||
| func TestCopyFields(t *testing.T) { | ||
|
|
||
| var tests = map[string]struct { | ||
| FromTo fromTo | ||
| Input common.MapStr | ||
| Expected common.MapStr | ||
| }{ | ||
| "copy string from message to message_copied": { | ||
| FromTo: fromTo{ | ||
| From: "message", | ||
| To: "message_copied", | ||
| }, | ||
| Input: common.MapStr{ | ||
| "message": "please copy this line", | ||
| }, | ||
| Expected: common.MapStr{ | ||
| "message": "please copy this line", | ||
| "message_copied": "please copy this line", | ||
| }, | ||
| }, | ||
| "copy string from nested key nested.message to top level field message_copied": { | ||
| FromTo: fromTo{ | ||
| From: "nested.message", | ||
| To: "message_copied", | ||
| }, | ||
| Input: common.MapStr{ | ||
| "nested": common.MapStr{ | ||
| "message": "please copy this line", | ||
| }, | ||
| }, | ||
| Expected: common.MapStr{ | ||
| "nested": common.MapStr{ | ||
| "message": "please copy this line", | ||
| }, | ||
| "message_copied": "please copy this line", | ||
| }, | ||
| }, | ||
| "copy string from fieldname with dot to message_copied": { | ||
| FromTo: fromTo{ | ||
| From: "dotted.message", | ||
| To: "message_copied", | ||
| }, | ||
| Input: common.MapStr{ | ||
| "dotted.message": "please copy this line", | ||
| }, | ||
| Expected: common.MapStr{ | ||
| "dotted.message": "please copy this line", | ||
| "message_copied": "please copy this line", | ||
| }, | ||
| }, | ||
| "copy number from fieldname with dot to dotted message.copied": { | ||
| FromTo: fromTo{ | ||
| From: "message.original", | ||
| To: "message.copied", | ||
| }, | ||
| Input: common.MapStr{ | ||
| "message.original": 42, | ||
| }, | ||
| Expected: common.MapStr{ | ||
| "message.original": 42, | ||
| "message": common.MapStr{ | ||
| "copied": 42, | ||
| }, | ||
| }, | ||
| }, | ||
| "copy number from hierarchical message.original to top level message which fails": { | ||
| FromTo: fromTo{ | ||
| From: "message.original", | ||
| To: "message", | ||
| }, | ||
| Input: common.MapStr{ | ||
| "message": common.MapStr{ | ||
| "original": 42, | ||
| }, | ||
| }, | ||
| Expected: common.MapStr{ | ||
| "message": common.MapStr{ | ||
| "original": 42, | ||
| }, | ||
| }, | ||
| }, | ||
| "copy number from hierarchical message.original to top level message": { | ||
| FromTo: fromTo{ | ||
| From: "message.original", | ||
| To: "message", | ||
| }, | ||
| Input: common.MapStr{ | ||
| "message.original": 42, | ||
| }, | ||
| Expected: common.MapStr{ | ||
| "message.original": 42, | ||
| "message": 42, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| for name, test := range tests { | ||
| t.Run(name, func(t *testing.T) { | ||
| p := copyFields{ | ||
| copyFieldsConfig{ | ||
| Fields: []fromTo{ | ||
| test.FromTo, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| event := &beat.Event{ | ||
| Fields: test.Input, | ||
| } | ||
|
|
||
| newEvent, err := p.Run(event) | ||
| assert.NoError(t, err) | ||
|
|
||
| assert.Equal(t, test.Expected, newEvent.Fields) | ||
| }) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.