Setup: Add default pipeline to templates

auditbeat/auditbeat.reference.yml

@@ -1066,6 +1066,18 @@ setup.template.settings:
   #_source:
     #enabled: false
 
+# Set to false to disable using a final pipeline.
+#setup.template.final_pipeline.enabled: true
+
+# Overwrite any existing final pipeline.
+#setup.template.final_pipeline.overwrite: false
+
+# Specify a custom name for the final pipeline.
+#setup.template.final_pipeline.name: "%{[agent.name]}-%{[agent.version]}"
+
+# Load a custom final pipeline from a file.
+#setup.template.final_pipeline.file:
+
 #============================== Setup ILM =====================================
 
 # Configure index lifecycle management (ILM). These settings create a write

filebeat/filebeat.reference.yml

@@ -1763,6 +1763,18 @@ setup.template.settings:
   #_source:
     #enabled: false
 
+# Set to false to disable using a final pipeline.
+#setup.template.final_pipeline.enabled: true
+
+# Overwrite any existing final pipeline.
+#setup.template.final_pipeline.overwrite: false
+
+# Specify a custom name for the final pipeline.
+#setup.template.final_pipeline.name: "%{[agent.name]}-%{[agent.version]}"
+
+# Load a custom final pipeline from a file.
+#setup.template.final_pipeline.file:
+
 #============================== Setup ILM =====================================
 
 # Configure index lifecycle management (ILM). These settings create a write

filebeat/fileset/fileset.go

@@ -24,7 +24,6 @@ package fileset
 
 import (
 	"bytes"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"io/ioutil"
@@ -35,10 +34,10 @@ import (
 	"text/template"
 
 	errw "github.com/pkg/errors"
-	"gopkg.in/yaml.v2"
 
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/cfgwarn"
+	commonP "github.com/elastic/beats/libbeat/common/pipeline"
 	"github.com/elastic/beats/libbeat/logp"
 )
 
@@ -414,7 +413,8 @@ func (fs *Fileset) GetPipelines(esVersion common.Version) (pipelines []pipeline,
 			return nil, fmt.Errorf("Error expanding vars on the ingest pipeline path: %v", err)
 		}
 
-		strContents, err := ioutil.ReadFile(filepath.Join(fs.modulePath, fs.name, path))
+		filePath := filepath.Join(fs.modulePath, fs.name, path)
+		strContents, err := ioutil.ReadFile(filePath)
 		if err != nil {
 			return nil, fmt.Errorf("Error reading pipeline file %s: %v", path, err)
 		}
@@ -424,23 +424,9 @@ func (fs *Fileset) GetPipelines(esVersion common.Version) (pipelines []pipeline,
 			return nil, fmt.Errorf("Error interpreting the template of the ingest pipeline: %v", err)
 		}
 
-		var content map[string]interface{}
-		switch extension := strings.ToLower(filepath.Ext(path)); extension {
-		case ".json":
-			if err = json.Unmarshal([]byte(encodedString), &content); err != nil {
-				return nil, fmt.Errorf("Error JSON decoding the pipeline file: %s: %v", path, err)
-			}
-		case ".yaml", ".yml":
-			if err = yaml.Unmarshal([]byte(encodedString), &content); err != nil {
-				return nil, fmt.Errorf("Error YAML decoding the pipeline file: %s: %v", path, err)
-			}
-			newContent, err := fixYAMLMaps(content)
-			if err != nil {
-				return nil, fmt.Errorf("Failed to sanitize the YAML pipeline file: %s: %v", path, err)
-			}
-			content = newContent.(map[string]interface{})
-		default:
-			return nil, fmt.Errorf("Unsupported extension '%s' for pipeline file: %s", extension, path)
+		content, err := commonP.UnmarshalPipeline(filePath, []byte(encodedString))
+		if err != nil {
+			return nil, fmt.Errorf("could not unmarshal pipeline file '%s'. Error: %v", filePath, err)
 		}
 
 		pipelineID := fs.pipelineIDs[idx]
@@ -455,40 +441,6 @@ func (fs *Fileset) GetPipelines(esVersion common.Version) (pipelines []pipeline,
 	return pipelines, nil
 }
 
-// This function recursively converts maps with interface{} keys, as returned by
-// yaml.Unmarshal, to maps of string keys, as expected by the json encoder
-// that will be used when delivering the pipeline to Elasticsearch.
-// Will return an error when something other than a string is used as a key.
-func fixYAMLMaps(elem interface{}) (_ interface{}, err error) {
-	switch v := elem.(type) {
-	case map[interface{}]interface{}:
-		result := make(map[string]interface{}, len(v))
-		for key, value := range v {
-			keyS, ok := key.(string)
-			if !ok {
-				return nil, fmt.Errorf("key '%v' is not string but %T", key, key)
-			}
-			if result[keyS], err = fixYAMLMaps(value); err != nil {
-				return nil, err
-			}
-		}
-		return result, nil
-	case map[string]interface{}:
-		for key, value := range v {
-			if v[key], err = fixYAMLMaps(value); err != nil {
-				return nil, err
-			}
-		}
-	case []interface{}:
-		for idx, value := range v {
-			if v[idx], err = fixYAMLMaps(value); err != nil {
-				return nil, err
-			}
-		}
-	}
-	return elem, nil
-}
-
 // formatPipelineID generates the ID to be used for the pipeline ID in Elasticsearch
 func formatPipelineID(module, fileset, path, beatVersion string) string {
 	return fmt.Sprintf("filebeat-%s-%s-%s-%s", beatVersion, module, fileset, removeExt(filepath.Base(path)))

heartbeat/heartbeat.reference.yml

@@ -1210,6 +1210,18 @@ setup.template.settings:
   #_source:
     #enabled: false
 
+# Set to false to disable using a final pipeline.
+#setup.template.final_pipeline.enabled: true
+
+# Overwrite any existing final pipeline.
+#setup.template.final_pipeline.overwrite: false
+
+# Specify a custom name for the final pipeline.
+#setup.template.final_pipeline.name: "%{[agent.name]}-%{[agent.version]}"
+
+# Load a custom final pipeline from a file.
+#setup.template.final_pipeline.file:
+
 #============================== Setup ILM =====================================
 
 # Configure index lifecycle management (ILM). These settings create a write

journalbeat/journalbeat.reference.yml

@@ -1004,6 +1004,18 @@ setup.template.settings:
   #_source:
     #enabled: false
 
+# Set to false to disable using a final pipeline.
+#setup.template.final_pipeline.enabled: true
+
+# Overwrite any existing final pipeline.
+#setup.template.final_pipeline.overwrite: false
+
+# Specify a custom name for the final pipeline.
+#setup.template.final_pipeline.name: "%{[agent.name]}-%{[agent.version]}"
+
+# Load a custom final pipeline from a file.
+#setup.template.final_pipeline.file:
+
 #============================== Setup ILM =====================================
 
 # Configure index lifecycle management (ILM). These settings create a write

libbeat/_meta/config.reference.yml.tmpl

@@ -947,6 +947,18 @@ setup.template.settings:
   #_source:
     #enabled: false
 
+# Set to false to disable using a final pipeline.
+#setup.template.final_pipeline.enabled: true
+
+# Overwrite any existing final pipeline.
+#setup.template.final_pipeline.overwrite: false
+
+# Specify a custom name for the final pipeline.
+#setup.template.final_pipeline.name: "%{[agent.name]}-%{[agent.version]}"
+
+# Load a custom final pipeline from a file.
+#setup.template.final_pipeline.file:
+
 #============================== Setup ILM =====================================
 
 # Configure index lifecycle management (ILM). These settings create a write

libbeat/cmd/export.go

@@ -35,6 +35,7 @@ func genExportCmd(settings instance.Settings) *cobra.Command {
 	exportCmd.AddCommand(export.GenIndexPatternConfigCmd(settings))
 	exportCmd.AddCommand(export.GenDashboardCmd(settings))
 	exportCmd.AddCommand(export.GenGetILMPolicyCmd(settings))
+	exportCmd.AddCommand(export.GenExportFinalPipelineCmd(settings))
 
 	return exportCmd
 }

libbeat/cmd/export/final_pipeline.go

@@ -0,0 +1,53 @@
+// 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 export
+
+import (
+	"github.com/spf13/cobra"
+
+	"github.com/elastic/beats/libbeat/cmd/instance"
+	"github.com/elastic/beats/libbeat/idxmgmt"
+)
+
+// GenExportFinalPipelineCmd writes out the final ingest pipeline.
+func GenExportFinalPipelineCmd(settings instance.Settings) *cobra.Command {
+	genTemplateConfigCmd := &cobra.Command{
+		Use:   "final-pipeline",
+		Short: "Export Elasticsearch final ingest pipeline to stdout",
+		Run: func(cmd *cobra.Command, args []string) {
+			version, _ := cmd.Flags().GetString("es.version")
+			dir, _ := cmd.Flags().GetString("dir")
+
+			b, err := instance.NewInitializedBeat(settings)
+			if err != nil {
+				fatalfInitCmd(err)
+			}
+
+			clientHandler := idxmgmt.NewFileClientHandler(newIdxmgmtClient(dir, version))
+			idxManager := b.IdxSupporter.Manager(clientHandler, idxmgmt.BeatsAssets(b.Fields))
+			if err := idxManager.Setup(idxmgmt.LoadModeDisabled, idxmgmt.LoadModeDisabled, idxmgmt.LoadModeForce); err != nil {
+				fatalf("Error exporting final-pipeline: %+v.", err)
+			}
+		},
+	}
+
+	genTemplateConfigCmd.Flags().String("es.version", settings.Version, "Elasticsearch version")
+	genTemplateConfigCmd.Flags().String("dir", "", "Specify directory to write pipeline file to. By default it is printed to stdout.")
+
+	return genTemplateConfigCmd
+}

libbeat/cmd/export/ilm_policy.go

@@ -44,7 +44,7 @@ func GenGetILMPolicyCmd(settings instance.Settings) *cobra.Command {
 
 			clientHandler := idxmgmt.NewFileClientHandler(newIdxmgmtClient(dir, version))
 			idxManager := b.IdxSupporter.Manager(clientHandler, idxmgmt.BeatsAssets(b.Fields))
-			if err := idxManager.Setup(idxmgmt.LoadModeDisabled, idxmgmt.LoadModeForce); err != nil {
+			if err := idxManager.Setup(idxmgmt.LoadModeDisabled, idxmgmt.LoadModeForce, idxmgmt.LoadModeDisabled); err != nil {
 				fatalf("Error exporting ilm-policy: %+v.", err)
 			}
 		},

libbeat/cmd/export/template.go

@@ -49,7 +49,7 @@ func GenTemplateConfigCmd(settings instance.Settings) *cobra.Command {
 
 			clientHandler := idxmgmt.NewFileClientHandler(newIdxmgmtClient(dir, version))
 			idxManager := b.IdxSupporter.Manager(clientHandler, idxmgmt.BeatsAssets(b.Fields))
-			if err := idxManager.Setup(idxmgmt.LoadModeForce, idxmgmt.LoadModeDisabled); err != nil {
+			if err := idxManager.Setup(idxmgmt.LoadModeForce, idxmgmt.LoadModeDisabled, idxmgmt.LoadModeDisabled); err != nil {
 				fatalf("Error exporting template: %+v.", err)
 			}
 		},

libbeat/cmd/instance/beat.go

@@ -501,18 +501,22 @@ func (b *Beat) Setup(settings Settings, bt beat.Creator, setup SetupSettings) er
 				return err
 			}
 
-			var loadTemplate, loadILM = idxmgmt.LoadModeUnset, idxmgmt.LoadModeUnset
+			var loadTemplate, loadILM, loadFinalPipeline = idxmgmt.LoadModeUnset, idxmgmt.LoadModeUnset, idxmgmt.LoadModeUnset
 			if setup.IndexManagement || setup.Template {
 				loadTemplate = idxmgmt.LoadModeOverwrite
 			}
 			if setup.IndexManagement || setup.ILMPolicy {
 				loadILM = idxmgmt.LoadModeEnabled
 			}
+			if setup.IndexManagement {
+				loadFinalPipeline = idxmgmt.LoadModeOverwrite
+			}
+
 			m := b.IdxSupporter.Manager(idxmgmt.NewESClientHandler(esClient), idxmgmt.BeatsAssets(b.Fields))
 			if ok, warn := m.VerifySetup(loadTemplate, loadILM); !ok {
 				fmt.Println(warn)
 			}
-			if err = m.Setup(loadTemplate, loadILM); err != nil {
+			if err = m.Setup(loadTemplate, loadILM, loadFinalPipeline); err != nil {
 				return err
 			}
 			fmt.Println("Index setup finished.")
@@ -801,7 +805,7 @@ func (b *Beat) registerESIndexManagement() error {
 func (b *Beat) indexSetupCallback() elasticsearch.ConnectCallback {
 	return func(esClient *elasticsearch.Client) error {
 		m := b.IdxSupporter.Manager(idxmgmt.NewESClientHandler(esClient), idxmgmt.BeatsAssets(b.Fields))
-		return m.Setup(idxmgmt.LoadModeEnabled, idxmgmt.LoadModeEnabled)
+		return m.Setup(idxmgmt.LoadModeEnabled, idxmgmt.LoadModeEnabled, idxmgmt.LoadModeEnabled)
 	}
 }
 

libbeat/common/fmtstr/formatevents.go

@@ -460,3 +460,20 @@ func tryConvString(v interface{}) (string, error) {
 		return "", errConvertString
 	}
 }
+
+// EventFormat takes an event format string and an event and returns
+// the expanded string. Returns error if the underlying parsing or
+// compilation fails.
+func EventFormat(format string, event *beat.Event) (string, error) {
+	formatter, err := CompileEvent(format)
+	if err != nil {
+		return "", err
+	}
+
+	expanded, err := formatter.Run(event)
+	if err != nil {
+		return "", err
+	}
+
+	return expanded, nil
+}