[Filebeat] Select output index based on the source input

filebeat/channel/wrapped.go

@@ -0,0 +1,71 @@
+package channel
+
+import (
+	"github.com/elastic/beats/libbeat/beat"
+	"github.com/elastic/beats/libbeat/common"
+)
+
+// A wrapper around a generic Outleter that applies the given transformation to
+// incoming events before forwarding them.
+type wrappedOutlet struct {
+	outlet         Outleter
+	eventTransform func(beat.Event)
+}
+
+// WrapOutlet takes an Outleter and an event transformation function and
+// returns an Outleter that applies that transformation before forwarding them.
+// The transformation operates in place (it modifies its input events).
+// The new Outleter uses the same underlying state, e.g. calling Close on the
+// wrapped Outleter will close the original as well. If this is not the intent,
+// call SubOutlet first.
+func WrapOutlet(outlet Outleter, eventTransform func(beat.Event)) Outleter {
+	return &wrappedOutlet{outlet: outlet, eventTransform: eventTransform}
+}
+
+func (o *wrappedOutlet) Close() error {
+	return o.outlet.Close()
+}
+
+func (o *wrappedOutlet) Done() <-chan struct{} {
+	return o.outlet.Done()
+}
+
+func (o *wrappedOutlet) OnEvent(event beat.Event) bool {
+	// Mutate the event then pass it on.
+	o.eventTransform(event)
+	return o.outlet.OnEvent(event)
+}
+
+// A wrapper around a generic Outleter that produces Outleters that apply the
+// given transformation to incoming events before sending them.
+type wrappedConnector struct {
+	connector      Connector
+	eventTransform func(beat.Event)
+}
+
+func (c *wrappedConnector) Connect(conf *common.Config) (Outleter, error) {
+	outleter, err := c.connector.Connect(conf)
+	if err != nil {
+		return outleter, err
+	}
+	return WrapOutlet(outleter, c.eventTransform), nil
+}
+
+func (c *wrappedConnector) ConnectWith(
+	conf *common.Config, clientConf beat.ClientConfig,
+) (Outleter, error) {
+	outleter, err := c.connector.ConnectWith(conf, clientConf)
+	if err != nil {
+		return outleter, err
+	}
+	return WrapOutlet(outleter, c.eventTransform), nil
+}
+
+// WrapConnector takes a Connector and an event transformation function and
+// returns a new Connector whose generated Outleters apply the given
+// transformation to incoming events before forwarding them.
+func WrapConnector(
+	connector Connector, eventTransform func(beat.Event),
+) Connector {
+	return &wrappedConnector{connector: connector, eventTransform: eventTransform}
+}

filebeat/input/config.go

@@ -35,6 +35,7 @@ type inputConfig struct {
 	ScanFrequency time.Duration `config:"scan_frequency" validate:"min=0,nonzero"`
 	Type          string        `config:"type"`
 	InputType     string        `config:"input_type"`
+	Index         string        `config:"index"`
 }
 
 func (c *inputConfig) Validate() error {

filebeat/input/input.go

@@ -26,7 +26,9 @@ import (
 
 	"github.com/elastic/beats/filebeat/channel"
 	"github.com/elastic/beats/filebeat/input/file"
+	"github.com/elastic/beats/libbeat/beat"
 	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/common/fmtstr"
 	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/libbeat/monitoring"
 )
@@ -60,7 +62,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,
@@ -91,6 +93,21 @@ func New(
 		return input, err
 	}
 
+	// If there is an input-level index setting, pass it through to event.Meta
+	// via the input's Connector.
+	if input.config.Index != "" {
+		indexPattern, err := fmtstr.CompileEvent(input.config.Index)
+		if err != nil {
+			return input, err
+		}
+		connector = channel.WrapConnector(connector, func(event beat.Event) {
+			if event.Meta == nil {
+				event.Meta = common.MapStr{}
+			}
+			event.Meta["index-pattern"] = indexPattern
+		})
+	}
+
 	context := Context{
 		States:        states,
 		Done:          input.done,
@@ -99,7 +116,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
 	}

libbeat/idxmgmt/std.go

@@ -20,10 +20,12 @@ package idxmgmt
 import (
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/elastic/beats/libbeat/beat"
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/atomic"
+	"github.com/elastic/beats/libbeat/common/fmtstr"
 	"github.com/elastic/beats/libbeat/idxmgmt/ilm"
 	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/libbeat/outputs"
@@ -54,12 +56,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 +207,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 +327,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 +341,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,9 +366,56 @@ func getEventCustomIndex(evt *beat.Event) string {
 		}
 	}
 
+	// index-pattern expands the index as formatted text rather than a fixed
+	// or dated string. This was introduced to support the `Index` configuration
+	// field on Filebeat inputs, so it's essentially a workaround to communicate
+	// a specific output behavior to libbeat based on the originating input
+	// (which as implemented now is a Filebeat-specific concept).
+	// If we find ourselves adding additional backchannels like this in Meta,
+	// we should probably add a more formal (non-MapStr) configuration struct to
+	// beat.Event that inputs can use to modulate output behavior.
+	if tmp := evt.Meta["index-pattern"]; tmp != nil {
+		if pattern, ok := tmp.(*fmtstr.EventFormatString); ok {
+			idx, err := expandIndexPattern(pattern, evt.Timestamp, beatInfo)
+			if err == nil {
+				return idx
+			}
+			// Should we log a warning here somehow?
+		}
+	}
+
 	return ""
 }
 
+// Expand the given pattern string as a formatted text field with access to
+// the event's agent and timestamp.
+// This helper mimicks applyStaticFmtstr in ilm.go, creating a placeholder
+// event for the restricted set of fields we allow here. It might be worth
+// making this a shared helper function or otherwise specializing for this case.
+func expandIndexPattern(
+	pattern *fmtstr.EventFormatString, timestamp time.Time, info beat.Info,
+) (string, error) {
+	return pattern.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(),
+	})
+}
+
 func unpackTemplateConfig(cfg *common.Config) (config template.TemplateConfig, err error) {
 	config = template.DefaultConfig()
 	if cfg != nil {