[Filebeat] Select output index based on the source input

CHANGELOG.next.asciidoc

@@ -379,6 +379,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]
 
 *Heartbeat*
 - Add non-privileged icmp on linux and darwin(mac). {pull}13795[13795] {issue}11498[11498]

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 := buildProcessorList(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,53 @@ func (c *pipelineConnector) ConnectWith(cfg *common.Config, clientCfg beat.Clien
 	}
 	return outlet, nil
 }
+
+// buildProcessorList assembles the Processors for a pipelineConnector.
+func buildProcessorList(
+	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
+	}
+	procs.List = append(procs.List, userProcessors)
+	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)
+}

filebeat/channel/connector_test.go

@@ -0,0 +1,185 @@
+// 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/elastic/beats/libbeat/beat"
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/processors"
+	"github.com/elastic/beats/libbeat/processors/actions"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBuildProcessorList(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 := buildProcessorList(test.beatInfo, config, test.clientCfg)
+		if err != nil {
+			t.Errorf("[%s] %v", description, err)
+			continue
+		}
+		processedEvent, err := processors.Run(&test.event)
+		if err != nil {
+			t.Error(err)
+			continue
+		}
+		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)
+			}
+		}
+	}
+}
+
+// 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 buildProcessorList.
+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
+}

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,19 +59,21 @@ 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
 // connecting an input to the publisher pipeline.
 func NewOutletFactory(
 	done <-chan struct{},
 	wgEvents eventCounter,
+	beatInfo beat.Info,
 ) *OutletFactory {
 	o := &OutletFactory{
 		done:     done,
 		wgEvents: wgEvents,
+		beatInfo: beatInfo,
 	}
 
 	if wgEvents != nil {

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"`.

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
 	}