[Filebeat] Select output index based on the source input

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"
 )
 
@@ -51,24 +54,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 +95,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
 	client, err := c.pipeline.ConnectWith(clientCfg)
 	if err != nil {
 		return nil, err
@@ -117,3 +107,63 @@ 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
+}
+
+//////////////////////////////
+// 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
+}
+
+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,195 @@
+// 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 := []struct {
+		description    string
+		beatInfo       beat.Info
+		configStr      string
+		clientCfg      beat.ClientConfig
+		event          beat.Event
+		expectedFields map[string]string
+	}{
+		{
+			description: "Simple static index",
+			configStr:   "index: 'test'",
+			expectedFields: map[string]string{
+				"@metadata.raw-index": "test",
+			},
+		},
+		{
+			description: "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",
+			},
+		},
+		{
+			description: "Set index in ClientConfig",
+			clientCfg: beat.ClientConfig{
+				Processing: beat.ProcessingConfig{
+					Processor: makeProcessors(&setRawIndex{"clientCfgIndex"}),
+				},
+			},
+			expectedFields: map[string]string{
+				"@metadata.raw-index": "clientCfgIndex",
+			},
+		},
+		{
+			description: "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",
+			},
+		},
+		{
+			description: "Set field in input config",
+			configStr: `processors:
+- add_fields: {fields: {testField: inputConfig}}`,
+			expectedFields: map[string]string{
+				"fields.testField": "inputConfig",
+			},
+		},
+		{
+			description: "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",
+			},
+		},
+		{
+			description: "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 _, 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", test.description, err)
+			continue
+		}
+		processors, err := buildProcessorList(test.beatInfo, config, test.clientCfg)
+		if err != nil {
+			t.Errorf("[%s] %v", test.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", test.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", test.description, key)
+				continue
+			}
+			if fieldStr != value {
+				t.Errorf("[%s] Event field [%s]: expected [%s], got [%s]", test.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 {
@@ -56,19 +58,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/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
 	}

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