-
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.
Add source path matching to
add_docker_metadata processor (#4495)
* Add `extract_field` internal processor * Add `match_source` to `add_docker_metadata` processor This should be useful to enrich events coming from docker logs, as it will parse the container id from the source path. * Move common docker fields to libbeat * Fix some messages and tests
- Loading branch information
1 parent
5c12a5f
commit 6bc3d4b
Showing
20 changed files
with
572 additions
and
69 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
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,91 @@ | ||
| package actions | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "strings" | ||
|
|
||
| "github.com/elastic/beats/libbeat/common" | ||
| "github.com/elastic/beats/libbeat/processors" | ||
| ) | ||
|
|
||
| type extract_field struct { | ||
| Field string | ||
| Separator string | ||
| Index int | ||
| Target string | ||
| } | ||
|
|
||
| /* | ||
| This one won't be registered (yet) | ||
| func init() { | ||
| processors.RegisterPlugin("extract_field", | ||
| configChecked(NewExtractField, | ||
| requireFields("field", "separator", "index", "target"), | ||
| allowedFields("field", "separator", "index", "target", "when"))) | ||
| } | ||
| */ | ||
|
|
||
| func NewExtractField(c common.Config) (processors.Processor, error) { | ||
| config := struct { | ||
| Field string `config:"field"` | ||
| Separator string `config:"separator"` | ||
| Index int `config:"index"` | ||
| Target string `config:"target"` | ||
| }{} | ||
| err := c.Unpack(&config) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("fail to unpack the extract_field configuration: %s", err) | ||
| } | ||
|
|
||
| /* remove read only fields */ | ||
| for _, readOnly := range processors.MandatoryExportedFields { | ||
| if config.Field == readOnly { | ||
| return nil, fmt.Errorf("%s is a read only field, cannot override", readOnly) | ||
| } | ||
| } | ||
|
|
||
| f := extract_field{ | ||
| Field: config.Field, | ||
| Separator: config.Separator, | ||
| Index: config.Index, | ||
| Target: config.Target, | ||
| } | ||
| return f, nil | ||
| } | ||
|
|
||
| func (f extract_field) Run(event common.MapStr) (common.MapStr, error) { | ||
| fieldValue, err := event.GetValue(f.Field) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("error getting field '%s' from event", f.Field) | ||
| } | ||
|
|
||
| value, ok := fieldValue.(string) | ||
| if !ok { | ||
| return nil, fmt.Errorf("could not get a string from field '%s'", f.Field) | ||
| } | ||
|
|
||
| parts := strings.Split(value, f.Separator) | ||
| parts = deleteEmpty(parts) | ||
| if len(parts) < f.Index+1 { | ||
| return nil, fmt.Errorf("index is out of range for field '%s'", f.Field) | ||
| } | ||
|
|
||
| event.Put(f.Target, parts[f.Index]) | ||
|
|
||
| return event, nil | ||
| } | ||
|
|
||
| func (f extract_field) String() string { | ||
| return "extract_field=" + f.Target | ||
| } | ||
|
|
||
| func deleteEmpty(s []string) []string { | ||
| var r []string | ||
| for _, str := range s { | ||
| if str != "" { | ||
| r = append(r, str) | ||
| } | ||
| } | ||
| return r | ||
| } |
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,90 @@ | ||
| package actions | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/elastic/beats/libbeat/common" | ||
| "github.com/elastic/beats/libbeat/logp" | ||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| func TestCommonPaths(t *testing.T) { | ||
| var tests = []struct { | ||
| Value, Field, Separator, Target, Result string | ||
| Index int | ||
| }{ | ||
| // Common docker case | ||
| { | ||
| Value: "/var/lib/docker/containers/f1510836197d7c34da22cf796dba5640f87c04de5c95cf0adc11b85f1e1c1528/f1510836197d7c34da22cf796dba5640f87c04de5c95cf0adc11b85f1e1c1528-json.log", | ||
| Field: "source", | ||
| Separator: "/", | ||
| Target: "docker.container.id", | ||
| Index: 4, | ||
| Result: "f1510836197d7c34da22cf796dba5640f87c04de5c95cf0adc11b85f1e1c1528", | ||
| }, | ||
| { | ||
| Value: "/var/lib/foo/bar", | ||
| Field: "other_field", | ||
| Separator: "/", | ||
| Target: "destination", | ||
| Index: 3, | ||
| Result: "bar", | ||
| }, | ||
| { | ||
| Value: "-var-lib-foo-bar", | ||
| Field: "source", | ||
| Separator: "-", | ||
| Target: "destination", | ||
| Index: 2, | ||
| Result: "foo", | ||
| }, | ||
| { | ||
| Value: "*var*lib*foo*bar", | ||
| Field: "source", | ||
| Separator: "*", | ||
| Target: "destination", | ||
| Index: 0, | ||
| Result: "var", | ||
| }, | ||
| } | ||
|
|
||
| for _, test := range tests { | ||
| var testConfig, _ = common.NewConfigFrom(map[string]interface{}{ | ||
| "field": test.Field, | ||
| "separator": test.Separator, | ||
| "index": test.Index, | ||
| "target": test.Target, | ||
| }) | ||
|
|
||
| // Configure input to | ||
| input := common.MapStr{ | ||
| test.Field: test.Value, | ||
| } | ||
|
|
||
| actual := runExtractField(t, testConfig, input) | ||
|
|
||
| result, err := actual.GetValue(test.Target) | ||
| if err != nil { | ||
| t.Fatalf("could not get target field: %s", err) | ||
| } | ||
| assert.Equal(t, result.(string), test.Result) | ||
| } | ||
| } | ||
|
|
||
| func runExtractField(t *testing.T, config *common.Config, input common.MapStr) common.MapStr { | ||
| if testing.Verbose() { | ||
| logp.LogInit(logp.LOG_DEBUG, "", false, true, []string{"*"}) | ||
| } | ||
|
|
||
| p, err := NewExtractField(*config) | ||
| if err != nil { | ||
| t.Fatalf("error initializing extract_field: %s", err) | ||
| } | ||
|
|
||
| actual, err := p.Run(input) | ||
| if err != nil { | ||
| t.Fatalf("error running extract_field: %s", err) | ||
| } | ||
|
|
||
| return actual | ||
| } |
Oops, something went wrong.