diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc index 85de3cfa347..660eae4558b 100644 --- a/CHANGELOG.next.asciidoc +++ b/CHANGELOG.next.asciidoc @@ -389,6 +389,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d - Add Kibana Dashboard for MISP module. {pull}14147[14147] - Add JSON options to autodiscover hints {pull}14208[14208] - Add more filesets to Zeek module. {pull}14150[14150] +- Add `index` option to all inputs to directly set a per-input index value. {pull}14010[14010] - Remove beta flag for some filebeat modules. {pull}14374[14374] *Heartbeat* diff --git a/filebeat/beater/filebeat.go b/filebeat/beater/filebeat.go index c30203dea16..76e7e8b12f8 100644 --- a/filebeat/beater/filebeat.go +++ b/filebeat/beater/filebeat.go @@ -326,7 +326,7 @@ func (fb *Filebeat) Run(b *beat.Beat) error { outDone := make(chan struct{}) // outDone closes down all active pipeline connections crawler, err := crawler.New( - channel.NewOutletFactory(outDone, wgEvents).Create, + channel.NewOutletFactory(outDone, wgEvents, b.Info).Create, config.Inputs, b.Info.Version, fb.done, diff --git a/filebeat/channel/connector.go b/filebeat/channel/connector.go index 3d584df38f5..ebd5983a3f7 100644 --- a/filebeat/channel/connector.go +++ b/filebeat/channel/connector.go @@ -18,8 +18,11 @@ package channel import ( + "fmt" + "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" ) @@ -31,6 +34,14 @@ type pipelineConnector struct { pipeline beat.Pipeline } +// addFormattedIndex is a Processor to set an event's "raw_index" metadata field +// with a given TimestampFormatString. The elasticsearch output interprets +// that field as specifying the (raw string) index the event should be sent to; +// in other outputs it is just included in the metadata. +type addFormattedIndex struct { + formatString *fmtstr.TimestampFormatString +} + // Connect passes the cfg and the zero value of beat.ClientConfig to the underlying function. func (fn ConnectorFunc) Connect(cfg *common.Config) (Outleter, error) { return fn(cfg, beat.ClientConfig{}) @@ -51,24 +62,11 @@ func (c *pipelineConnector) ConnectWith(cfg *common.Config, clientCfg beat.Clien return nil, err } - var err error - var userProcessors beat.ProcessorList - - userProcessors, err = processors.New(config.Processors) + procs, err := processorsForConfig(c.parent.beatInfo, config, clientCfg) if err != nil { return nil, err } - if lst := clientCfg.Processing.Processor; lst != nil { - if len(userProcessors.All()) == 0 { - userProcessors = lst - } else if orig := lst.All(); len(orig) > 0 { - newLst := processors.NewList(nil) - newLst.List = append(newLst.List, lst, userProcessors) - userProcessors = newLst - } - } - setOptional := func(to common.MapStr, key string, value string) { if value != "" { to.Put(key, value) @@ -105,7 +103,7 @@ func (c *pipelineConnector) ConnectWith(cfg *common.Config, clientCfg beat.Clien clientCfg.Processing.EventMetadata = config.EventMetadata clientCfg.Processing.Meta = meta clientCfg.Processing.Fields = fields - clientCfg.Processing.Processor = userProcessors + clientCfg.Processing.Processor = procs clientCfg.Processing.KeepNull = config.KeepNull client, err := c.pipeline.ConnectWith(clientCfg) if err != nil { @@ -118,3 +116,64 @@ func (c *pipelineConnector) ConnectWith(cfg *common.Config, clientCfg beat.Clien } return outlet, nil } + +// processorsForConfig assembles the Processors for a pipelineConnector. +func processorsForConfig( + beatInfo beat.Info, config inputOutletConfig, clientCfg beat.ClientConfig, +) (*processors.Processors, error) { + procs := processors.NewList(nil) + + // Processor ordering is important: + // 1. Index configuration + 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 := &addFormattedIndex{timestampFormat} + procs.List = append(procs.List, indexProcessor) + } + + // 2. ClientConfig processors + if lst := clientCfg.Processing.Processor; lst != nil { + procs.List = append(procs.List, lst) + } + + // 3. User processors + userProcessors, err := processors.New(config.Processors) + if err != nil { + return nil, err + } + // Subtlety: it is important here that we append the individual elements of + // userProcessors, rather than userProcessors itself, even though + // userProcessors implements the processors.Processor interface. This is + // because the contents of what we return are later pulled out into a + // processing.group rather than a processors.Processors, and the two have + // different error semantics: processors.Processors aborts processing on + // any error, whereas processing.group only aborts on fatal errors. The + // latter is the most common behavior, and the one we are preserving here for + // backwards compatibility. + // We are unhappy about this and have plans to fix this inconsistency at a + // higher level, but for now we need to respect the existing semantics. + procs.List = append(procs.List, userProcessors.List...) + return procs, nil +} + +func (p *addFormattedIndex) Run(event *beat.Event) (*beat.Event, error) { + index, err := p.formatString.Run(event.Timestamp) + if err != nil { + return nil, err + } + + if event.Meta == nil { + event.Meta = common.MapStr{} + } + event.Meta["raw_index"] = index + return event, nil +} + +func (p *addFormattedIndex) String() string { + return fmt.Sprintf("add_index_pattern=%v", p.formatString) +} diff --git a/filebeat/channel/connector_test.go b/filebeat/channel/connector_test.go new file mode 100644 index 00000000000..4708a7e45a5 --- /dev/null +++ b/filebeat/channel/connector_test.go @@ -0,0 +1,213 @@ +// 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 channel + +import ( + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/assert" + + "github.com/elastic/beats/libbeat/beat" + "github.com/elastic/beats/libbeat/common" + "github.com/elastic/beats/libbeat/processors" + "github.com/elastic/beats/libbeat/processors/actions" +) + +func TestProcessorsForConfig(t *testing.T) { + testCases := map[string]struct { + beatInfo beat.Info + configStr string + clientCfg beat.ClientConfig + 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", + }, + }, + "Set index in ClientConfig": { + clientCfg: beat.ClientConfig{ + Processing: beat.ProcessingConfig{ + Processor: makeProcessors(&setRawIndex{"clientCfgIndex"}), + }, + }, + expectedFields: map[string]string{ + "@metadata.raw_index": "clientCfgIndex", + }, + }, + "ClientConfig processor runs after beat input Index": { + configStr: "index: 'test'", + clientCfg: beat.ClientConfig{ + Processing: beat.ProcessingConfig{ + Processor: makeProcessors(&setRawIndex{"clientCfgIndex"}), + }, + }, + expectedFields: map[string]string{ + "@metadata.raw_index": "clientCfgIndex", + }, + }, + "Set field in input config": { + configStr: `processors: [add_fields: {fields: {testField: inputConfig}}]`, + expectedFields: map[string]string{ + "fields.testField": "inputConfig", + }, + }, + "Set field in ClientConfig": { + clientCfg: beat.ClientConfig{ + Processing: beat.ProcessingConfig{ + Processor: makeProcessors(actions.NewAddFields(common.MapStr{ + "fields": common.MapStr{"testField": "clientConfig"}, + }, false)), + }, + }, + expectedFields: map[string]string{ + "fields.testField": "clientConfig", + }, + }, + "Input config processors run after ClientConfig": { + configStr: `processors: [add_fields: {fields: {testField: inputConfig}}]`, + clientCfg: beat.ClientConfig{ + Processing: beat.ProcessingConfig{ + Processor: makeProcessors(actions.NewAddFields(common.MapStr{ + "fields": common.MapStr{"testField": "clientConfig"}, + }, false)), + }, + }, + expectedFields: map[string]string{ + "fields.testField": "inputConfig", + }, + }, + } + for description, test := range testCases { + if test.event.Fields == nil { + test.event.Fields = common.MapStr{} + } + config, err := outletConfigFromString(test.configStr) + if err != nil { + t.Errorf("[%s] %v", description, err) + continue + } + processors, err := processorsForConfig(test.beatInfo, config, test.clientCfg) + 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) + } + } + } +} + +func TestProcessorsForConfigIsFlat(t *testing.T) { + // This test is regrettable, and exists because of inconsistencies in + // processor handling between processors.Processors and processing.group + // (which implements beat.ProcessorList) -- see processorsForConfig for + // details. The upshot is that, for now, if the input configuration specifies + // processors, they must be returned as direct children of the resulting + // processors.Processors (rather than being collected in additional tree + // structure). + // This test should be removed once we have a more consistent mechanism for + // collecting and running processors. + configStr := `processors: +- add_fields: {fields: {testField: value}} +- add_fields: {fields: {testField2: stuff}}` + config, err := outletConfigFromString(configStr) + if err != nil { + t.Fatal(err) + } + processors, err := processorsForConfig( + beat.Info{}, config, beat.ClientConfig{}) + if err != nil { + t.Fatal(err) + } + assert.Equal(t, 2, len(processors.List)) +} + +// setRawIndex is a bare-bones processor to set the raw_index field to a +// constant string in the event metadata. It is used to test order of operations +// for processorsForConfig. +type setRawIndex struct { + indexStr string +} + +func (p *setRawIndex) Run(event *beat.Event) (*beat.Event, error) { + if event.Meta == nil { + event.Meta = common.MapStr{} + } + event.Meta["raw_index"] = p.indexStr + return event, nil +} + +func (p *setRawIndex) String() string { + return fmt.Sprintf("set_raw_index=%v", p.indexStr) +} + +// Helper function to convert from YML input string to an unpacked +// inputOutletConfig +func outletConfigFromString(s string) (inputOutletConfig, error) { + config := inputOutletConfig{} + cfg, err := common.NewConfigFrom(s) + if err != nil { + return config, err + } + if err := cfg.Unpack(&config); err != nil { + return config, err + } + return config, nil +} + +// makeProcessors wraps one or more bare Processor objects in Processors. +func makeProcessors(procs ...processors.Processor) *processors.Processors { + procList := processors.NewList(nil) + procList.List = procs + return procList +} diff --git a/filebeat/channel/factory.go b/filebeat/channel/factory.go index 1fe09039bbe..d4373beec19 100644 --- a/filebeat/channel/factory.go +++ b/filebeat/channel/factory.go @@ -20,6 +20,7 @@ package channel 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" ) @@ -28,6 +29,7 @@ type OutletFactory struct { eventer beat.ClientEventer wgEvents eventCounter + beatInfo beat.Info } type eventCounter interface { @@ -57,8 +59,8 @@ type inputOutletConfig struct { Fileset string `config:"_fileset_name"` // hidden setting // Output meta data settings - Pipeline string `config:"pipeline"` // ES Ingest pipeline name - + Pipeline string `config:"pipeline"` // ES Ingest pipeline name + Index fmtstr.EventFormatString `config:"index"` // ES output index pattern } // NewOutletFactory creates a new outlet factory for @@ -66,10 +68,12 @@ type inputOutletConfig struct { func NewOutletFactory( done <-chan struct{}, wgEvents eventCounter, + beatInfo beat.Info, ) *OutletFactory { o := &OutletFactory{ done: done, wgEvents: wgEvents, + beatInfo: beatInfo, } if wgEvents != nil { diff --git a/filebeat/docs/inputs/input-common-options.asciidoc b/filebeat/docs/inputs/input-common-options.asciidoc index 6d09ff493b9..8e08a8074e4 100644 --- a/filebeat/docs/inputs/input-common-options.asciidoc +++ b/filebeat/docs/inputs/input-common-options.asciidoc @@ -64,7 +64,7 @@ If this option is set to true, the custom <<{beatname_lc}-input-{type}-fields,fields>> are stored as top-level fields in the output document instead of being grouped under a `fields` sub-dictionary. If the custom field names conflict with other field names added by {beatname_uc}, -then the custom fields overwrite the other fields. +then the custom fields overwrite the other fields. [float] ===== `processors` @@ -89,3 +89,15 @@ input is used. If this option is set to true, fields with `null` values will be published in the output document. By default, `keep_null` is set to `false`. + +[float] +===== `index` + +If present, this formatted string overrides the index for events from this input +(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 `"filebeat-myindex-2019.11.01"`. diff --git a/filebeat/input/input.go b/filebeat/input/input.go index 98eea51db8a..e1931d4e4bf 100644 --- a/filebeat/input/input.go +++ b/filebeat/input/input.go @@ -60,7 +60,7 @@ type Runner struct { // New instantiates a new Runner func New( conf *common.Config, - outlet channel.Connector, + connector channel.Connector, beatDone chan struct{}, states []file.State, dynFields *common.MapStrPointer, @@ -99,7 +99,7 @@ func New( Meta: nil, } var ipt Input - ipt, err = f(conf, outlet, context) + ipt, err = f(conf, connector, context) if err != nil { return input, err } diff --git a/filebeat/kafka.yml b/filebeat/kafka.yml new file mode 100644 index 00000000000..81842758971 --- /dev/null +++ b/filebeat/kafka.yml @@ -0,0 +1,8 @@ +version: '2.3' +services: + kafka: + build: ${ES_BEATS}/testing/environments/docker/kafka + expose: + - 9092 + environment: + - ADVERTISED_HOST=kafka diff --git a/libbeat/common/fmtstr/formattimestamp.go b/libbeat/common/fmtstr/formattimestamp.go new file mode 100644 index 00000000000..68c22d63649 --- /dev/null +++ b/libbeat/common/fmtstr/formattimestamp.go @@ -0,0 +1,80 @@ +// 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 fmtstr + +import ( + "time" + + "github.com/elastic/beats/libbeat/beat" + "github.com/elastic/beats/libbeat/common" +) + +// TimestampFormatString is a wrapper around EventFormatString for the +// common special case where the format expression should only have access to +// shared static fields (typically agent / version) and the event timestamp. +type TimestampFormatString struct { + eventFormatString *EventFormatString + fields common.MapStr +} + +// NewTimestampFormatString creates from the given event format string a +// TimestampFormatString that includes only the given static fields and +// a timestamp. +func NewTimestampFormatString( + eventFormatString *EventFormatString, staticFields common.MapStr, +) (*TimestampFormatString, error) { + return &TimestampFormatString{ + eventFormatString: eventFormatString, + fields: staticFields.Clone(), + }, nil +} + +// FieldsForBeat returns a common.MapStr with the given beat name and +// version assigned to their standard field names. +func FieldsForBeat(beat string, version string) common.MapStr { + return common.MapStr{ + // beat object was left in for backward compatibility reason for older configs. + "beat": common.MapStr{ + "name": beat, + "version": version, + }, + "agent": common.MapStr{ + "name": beat, + "version": version, + }, + // For the Beats that have an observer role + "observer": common.MapStr{ + "name": beat, + "version": version, + }, + } +} + +// Run executes the format string returning a new expanded string or an error +// if execution or event field expansion fails. +func (fs *TimestampFormatString) Run(timestamp time.Time) (string, error) { + placeholderEvent := &beat.Event{ + Fields: fs.fields, + Timestamp: timestamp, + } + return fs.eventFormatString.Run(placeholderEvent) +} + +func (fs *TimestampFormatString) String() string { + return fs.eventFormatString.expression +} diff --git a/libbeat/common/fmtstr/formattimestamp_test.go b/libbeat/common/fmtstr/formattimestamp_test.go new file mode 100644 index 00000000000..d194f597ad5 --- /dev/null +++ b/libbeat/common/fmtstr/formattimestamp_test.go @@ -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 fmtstr + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + + "github.com/elastic/beats/libbeat/common" +) + +func TestTimestampFormatString(t *testing.T) { + tests := []struct { + title string + format string + staticFields common.MapStr + timestamp time.Time + expected string + }{ + { + "empty string", + "", + nil, + time.Time{}, + "", + }, + { + "no fields configured", + "format string", + nil, + time.Time{}, + "format string", + }, + { + "expand field", + "%{[key]}", + common.MapStr{"key": "value"}, + time.Time{}, + "value", + }, + { + "expand with default", + "%{[key]:default}", + nil, + time.Time{}, + "default", + }, + { + "expand nested field", + "%{[nested.key]}", + common.MapStr{"nested": common.MapStr{"key": "value"}}, + time.Time{}, + "value", + }, + { + "test timestamp formatter", + "%{[key]}: %{+YYYY.MM.dd}", + common.MapStr{"key": "timestamp"}, + time.Date(2015, 5, 1, 20, 12, 34, 0, time.Local), + "timestamp: 2015.05.01", + }, + { + "test timestamp formatter", + "%{[@timestamp]}: %{+YYYY.MM.dd}", + common.MapStr{"key": "timestamp"}, + time.Date(2015, 5, 1, 20, 12, 34, 0, time.Local), + "2015-05-01T20:12:34.000Z: 2015.05.01", + }, + } + + for i, test := range tests { + t.Logf("test(%v): %v", i, test.title) + + efs, err := CompileEvent(test.format) + if err != nil { + t.Error(err) + continue + } + + fs, err := NewTimestampFormatString(efs, test.staticFields) + if err != nil { + t.Error(err) + continue + } + + actual, err := fs.Run(test.timestamp) + + assert.NoError(t, err) + assert.Equal(t, test.expected, actual) + } +} diff --git a/libbeat/idxmgmt/ilm/ilm.go b/libbeat/idxmgmt/ilm/ilm.go index 07c924214ec..84ca92c432e 100644 --- a/libbeat/idxmgmt/ilm/ilm.go +++ b/libbeat/idxmgmt/ilm/ilm.go @@ -142,23 +142,9 @@ func NoopSupport(_ *logp.Logger, info beat.Info, config *common.Config) (Support } func applyStaticFmtstr(info beat.Info, fmt *fmtstr.EventFormatString) (string, error) { - return fmt.Run(&beat.Event{ - Fields: common.MapStr{ - // beat object was left in for backward compatibility reason for older configs. - "beat": common.MapStr{ - "name": info.Beat, - "version": info.Version, - }, - "agent": common.MapStr{ - "name": info.Beat, - "version": info.Version, - }, - // For the Beats that have an observer role - "observer": common.MapStr{ - "name": info.Beat, - "version": info.Version, - }, - }, - Timestamp: time.Now(), - }) + return fmt.Run( + &beat.Event{ + Fields: fmtstr.FieldsForBeat(info.Beat, info.Version), + Timestamp: time.Now(), + }) } diff --git a/libbeat/idxmgmt/std.go b/libbeat/idxmgmt/std.go index 13d1a201fcf..f6951d729ad 100644 --- a/libbeat/idxmgmt/std.go +++ b/libbeat/idxmgmt/std.go @@ -54,12 +54,16 @@ type indexManager struct { assets Asseter } -type indexSelector outil.Selector +type indexSelector struct { + sel outil.Selector + beatInfo beat.Info +} type ilmIndexSelector struct { - index outil.Selector - alias outil.Selector - st *indexState + index outil.Selector + alias outil.Selector + st *indexState + beatInfo beat.Info } type componentType uint8 @@ -201,7 +205,7 @@ func (s *indexSupport) BuildSelector(cfg *common.Config) (outputs.IndexSelector, } if mode != ilm.ModeAuto { - return indexSelector(indexSel), nil + return indexSelector{indexSel, s.info}, nil } selCfg.SetString("index", -1, alias) @@ -321,7 +325,7 @@ func (m *indexManager) setupWithILM() (bool, error) { } func (s *ilmIndexSelector) Select(evt *beat.Event) (string, error) { - if idx := getEventCustomIndex(evt); idx != "" { + if idx := getEventCustomIndex(evt, s.beatInfo); idx != "" { return idx, nil } @@ -335,13 +339,13 @@ func (s *ilmIndexSelector) Select(evt *beat.Event) (string, error) { } func (s indexSelector) Select(evt *beat.Event) (string, error) { - if idx := getEventCustomIndex(evt); idx != "" { + if idx := getEventCustomIndex(evt, s.beatInfo); idx != "" { return idx, nil } - return outil.Selector(s).Select(evt) + return s.sel.Select(evt) } -func getEventCustomIndex(evt *beat.Event) string { +func getEventCustomIndex(evt *beat.Event, beatInfo beat.Info) string { if len(evt.Meta) == 0 { return "" } @@ -360,6 +364,16 @@ func getEventCustomIndex(evt *beat.Event) string { } } + // This is functionally identical to Meta["alias"], returning the overriding + // metadata as the index name if present. It is currently used by Filebeat + // to send the index for particular inputs to formatted string templates, + // which are then expanded by a processor to the "raw_index" field. + if tmp := evt.Meta["raw_index"]; tmp != nil { + if idx, ok := tmp.(string); ok { + return idx + } + } + return "" }