Skip to content

Commit

Permalink
[metricbeat] [auditbeat] Add formatted index option to metricbeat / a…
Browse files Browse the repository at this point in the history
…uditbeat modules (#15100) (#15130)
  • Loading branch information
faec authored Dec 17, 2019
1 parent 23eb8ac commit a73fa46
Show file tree
Hide file tree
Showing 7 changed files with 167 additions and 11 deletions.
1 change: 1 addition & 0 deletions CHANGELOG.next.asciidoc
Original file line number Diff line number Diff line change
Expand Up @@ -531,6 +531,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
- Add overview dashboard for AWS SNS module {pull}14977[14977]
- Add STAN Metricbeat module. {pull}14839[14839]
- Add a `system/service` metricset for systemd data. {pull}14206[14206]
- Add `index` option to all modules to specify a module-specific output index. {pull}15100[15100]

*Packetbeat*

Expand Down
6 changes: 3 additions & 3 deletions metricbeat/beater/metricbeat.go
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,7 @@ func newMetricbeat(b *beat.Beat, c *common.Config, options ...Option) (*Metricbe

failed := false

connector, err := module.NewConnector(b.Publisher, moduleCfg, nil)
connector, err := module.NewConnector(b.Info, b.Publisher, moduleCfg, nil)
if err != nil {
errs = append(errs, err)
failed = true
Expand Down Expand Up @@ -201,7 +201,7 @@ func newMetricbeat(b *beat.Beat, c *common.Config, options ...Option) (*Metricbe

if config.Autodiscover != nil {
var err error
factory := module.NewFactory(metricbeat.moduleOptions...)
factory := module.NewFactory(b.Info, metricbeat.moduleOptions...)
adapter := autodiscover.NewFactoryAdapter(factory)
metricbeat.autodiscover, err = autodiscover.NewAutodiscover("metricbeat", b.Publisher, adapter, config.Autodiscover)
if err != nil {
Expand Down Expand Up @@ -238,7 +238,7 @@ func (bt *Metricbeat) Run(b *beat.Beat) error {
}

// Centrally managed modules
factory := module.NewFactory(bt.moduleOptions...)
factory := module.NewFactory(b.Info, bt.moduleOptions...)
modules := cfgfile.NewRunnerList(management.DebugK, factory, b.Publisher)
reload.Register.MustRegisterList(b.Info.Beat+".modules", modules)
wg.Add(1)
Expand Down
14 changes: 13 additions & 1 deletion metricbeat/docs/metricbeat-options.asciidoc
Original file line number Diff line number Diff line change
Expand Up @@ -214,6 +214,18 @@ A list of processors to apply to the data generated by the metricset.
See <<filtering-and-enhancing-data>> for information about specifying
processors in your config.

[float]
==== `index`

If present, this formatted string overrides the index for events from this
module (for elasticsearch outputs), or sets the `raw_index` field of the event's
metadata (for other outputs). This string can only refer to the agent name and
version and the event timestamp; for access to dynamic fields, use
`output.elasticsearch.index` or a processor.

Example value: `"%{[agent.name]}-myindex-%{+yyyy.MM.dd}"` might
expand to `"metricbeat-myindex-2019.12.13"`.

[float]
==== `keep_null`

Expand Down Expand Up @@ -289,7 +301,7 @@ as the first segment in the HTTP URI path.
[float]
==== `query`

An optional value to pass common query params in YAML. Instead of setting the query params
An optional value to pass common query params in YAML. Instead of setting the query params
within hosts values using the syntax `?key=value&key2&value2`, you can set it here like this:

[source,yaml]
Expand Down
39 changes: 37 additions & 2 deletions metricbeat/mb/module/connector.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,9 @@ package module
import (
"github.com/elastic/beats/libbeat/beat"
"github.com/elastic/beats/libbeat/common"
"github.com/elastic/beats/libbeat/common/fmtstr"
"github.com/elastic/beats/libbeat/processors"
"github.com/elastic/beats/libbeat/processors/add_formatted_index"
)

// Connector configures and establishes a beat.Client for publishing events
Expand All @@ -36,20 +38,25 @@ type Connector struct {

type connectorConfig struct {
Processors processors.PluginConfig `config:"processors"`
// ES output index pattern
Index fmtstr.EventFormatString `config:"index"`

// KeepNull determines whether published events will keep null values or omit them.
KeepNull bool `config:"keep_null"`

common.EventMetadata `config:",inline"` // Fields and tags to add to events.
}

func NewConnector(pipeline beat.Pipeline, c *common.Config, dynFields *common.MapStrPointer) (*Connector, error) {
func NewConnector(
beatInfo beat.Info, pipeline beat.Pipeline,
c *common.Config, dynFields *common.MapStrPointer,
) (*Connector, error) {
config := connectorConfig{}
if err := c.Unpack(&config); err != nil {
return nil, err
}

processors, err := processors.New(config.Processors)
processors, err := processorsForConfig(beatInfo, config)
if err != nil {
return nil, err
}
Expand All @@ -73,3 +80,31 @@ func (c *Connector) Connect() (beat.Client, error) {
},
})
}

// processorsForConfig assembles the Processors for a Connector.
func processorsForConfig(
beatInfo beat.Info, config connectorConfig,
) (*processors.Processors, error) {
procs := processors.NewList(nil)

// Processor order is important! The index processor, if present, must be
// added before the user processors.
if !config.Index.IsEmpty() {
staticFields := fmtstr.FieldsForBeat(beatInfo.Beat, beatInfo.Version)
timestampFormat, err :=
fmtstr.NewTimestampFormatString(&config.Index, staticFields)
if err != nil {
return nil, err
}
indexProcessor := add_formatted_index.New(timestampFormat)
procs.AddProcessor(indexProcessor)
}

userProcs, err := processors.New(config.Processors)
if err != nil {
return nil, err
}
procs.AddProcessors(*userProcs)

return procs, nil
}
106 changes: 106 additions & 0 deletions metricbeat/mb/module/connector_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,106 @@
// 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 module

import (
"testing"
"time"

"github.com/stretchr/testify/assert"

"github.com/elastic/beats/libbeat/beat"
"github.com/elastic/beats/libbeat/common"
)

func TestProcessorsForConfig(t *testing.T) {
testCases := map[string]struct {
beatInfo beat.Info
configStr string
event beat.Event
expectedFields map[string]string
}{
"Simple static index": {
configStr: "index: 'test'",
expectedFields: map[string]string{
"@metadata.raw_index": "test",
},
},
"Index with agent info + timestamp": {
beatInfo: beat.Info{Beat: "TestBeat", Version: "3.9.27"},
configStr: "index: 'beat-%{[agent.name]}-%{[agent.version]}-%{+yyyy.MM.dd}'",
event: beat.Event{Timestamp: time.Date(1999, time.December, 31, 23, 0, 0, 0, time.UTC)},
expectedFields: map[string]string{
"@metadata.raw_index": "beat-TestBeat-3.9.27-1999.12.31",
},
},
}
for description, test := range testCases {
if test.event.Fields == nil {
test.event.Fields = common.MapStr{}
}
config, err := connectorConfigFromString(test.configStr)
if err != nil {
t.Errorf("[%s] %v", description, err)
continue
}
processors, err := processorsForConfig(test.beatInfo, config)
if err != nil {
t.Errorf("[%s] %v", description, err)
continue
}
processedEvent, err := processors.Run(&test.event)
// We don't check if err != nil, because we are testing the final outcome
// of running the processors, including when some of them fail.
if processedEvent == nil {
t.Errorf("[%s] Unexpected fatal error running processors: %v\n",
description, err)
}
for key, value := range test.expectedFields {
field, err := processedEvent.GetValue(key)
if err != nil {
t.Errorf("[%s] Couldn't get field %s from event: %v", description, key, err)
continue
}
assert.Equal(t, field, value)
fieldStr, ok := field.(string)
if !ok {
// Note that requiring a string here is just to simplify the test setup,
// not a requirement of the underlying api.
t.Errorf("[%s] Field [%s] should be a string", description, key)
continue
}
if fieldStr != value {
t.Errorf("[%s] Event field [%s]: expected [%s], got [%s]", description, key, value, fieldStr)
}
}
}
}

// Helper function to convert from YML input string to an unpacked
// connectorConfig
func connectorConfigFromString(s string) (connectorConfig, error) {
config := connectorConfig{}
cfg, err := common.NewConfigFrom(s)
if err != nil {
return config, err
}
if err := cfg.Unpack(&config); err != nil {
return config, err
}
return config, nil
}
2 changes: 1 addition & 1 deletion metricbeat/mb/module/example_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -132,7 +132,7 @@ func ExampleRunner() {
return
}

connector, err := module.NewConnector(b.Publisher, config, nil)
connector, err := module.NewConnector(b.Info, b.Publisher, config, nil)
if err != nil {
return
}
Expand Down
10 changes: 6 additions & 4 deletions metricbeat/mb/module/factory.go
Original file line number Diff line number Diff line change
Expand Up @@ -29,21 +29,23 @@ import (
// Factory creates new Runner instances from configuration objects.
// It is used to register and reload modules.
type Factory struct {
options []Option
beatInfo beat.Info
options []Option
}

// NewFactory creates new Reloader instance for the given config
func NewFactory(options ...Option) *Factory {
func NewFactory(beatInfo beat.Info, options ...Option) *Factory {
return &Factory{
options: options,
beatInfo: beatInfo,
options: options,
}
}

// Create creates a new metricbeat module runner reporting events to the passed pipeline.
func (r *Factory) Create(p beat.Pipeline, c *common.Config, meta *common.MapStrPointer) (cfgfile.Runner, error) {
var errs multierror.Errors

connector, err := NewConnector(p, c, meta)
connector, err := NewConnector(r.beatInfo, p, c, meta)
if err != nil {
errs = append(errs, err)
}
Expand Down

0 comments on commit a73fa46

Please sign in to comment.