From c08b36b14b090248cd2c281fb702338c62a73ff3 Mon Sep 17 00:00:00 2001
From: Tudor Golubenco <tudor@packetbeat.com>
Date: Wed, 6 Jul 2016 17:55:37 +0200
Subject: [PATCH] Load template by default, depending on the version

This queries Elasticsearch for the version and choses between the
default template file (specified by the `template.path` option) and
the ES 2x version of the template file (specified by the new
`template.versions.2x.path` option).

There's also a change in the defaults. With all the template
options commented out, the code now loads the template. Previously,
the template loading was the default via the sample configs only.

I also did a small change in the output plugins API, now they receive
the beatname in a clean way.

Fixes #1740.
---
 CHANGELOG.asciidoc                            |  2 +
 filebeat/filebeat.full.yml                    | 20 ++++++--
 filebeat/filebeat.yml                         |  9 ----
 libbeat/_meta/config.full.yml                 | 20 ++++++--
 libbeat/_meta/config.yml                      |  9 ----
 libbeat/docs/outputconfig.asciidoc            | 31 ++++++++++--
 libbeat/outputs/console/console.go            |  2 +-
 .../outputs/elasticsearch/api_mock_test.go    | 10 ++--
 libbeat/outputs/elasticsearch/client.go       | 29 +++++++++--
 .../elasticsearch/client_integration_test.go  |  7 +--
 libbeat/outputs/elasticsearch/config.go       | 21 ++++++--
 libbeat/outputs/elasticsearch/output.go       | 49 +++++++++++++++----
 libbeat/outputs/elasticsearch/output_test.go  | 23 ++++-----
 libbeat/outputs/fileout/file.go               |  9 ++--
 libbeat/outputs/kafka/kafka.go                |  2 +-
 .../outputs/kafka/kafka_integration_test.go   |  2 +-
 libbeat/outputs/logstash/logstash.go          |  2 +-
 .../logstash/logstash_integration_test.go     | 15 +++---
 libbeat/outputs/logstash/logstash_test.go     |  2 +-
 libbeat/outputs/outputs.go                    |  4 +-
 libbeat/outputs/redis/redis.go                |  2 +-
 .../outputs/redis/redis_integration_test.go   |  2 +-
 libbeat/tests/system/test_base.py             |  4 +-
 metricbeat/metricbeat.full.yml                | 20 ++++++--
 metricbeat/metricbeat.yml                     |  9 ----
 packetbeat/packetbeat.full.yml                | 20 ++++++--
 packetbeat/packetbeat.yml                     |  9 ----
 winlogbeat/winlogbeat.full.yml                | 20 ++++++--
 winlogbeat/winlogbeat.yml                     |  9 ----
 29 files changed, 237 insertions(+), 126 deletions(-)

diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index d5941259ce6..13b1def970d 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -15,6 +15,7 @@ https://github.com/elastic/beats/compare/v5.0.0-alpha4...master[Check the HEAD d
 *Affecting all Beats*
 - Rename the `filters` section to `processors`. {pull}1944[1944]
 - Introduce the condition with `when` in the processor configuration. {pull}1949[1949]
+- The Elasticsearch template is now loaded by default. {pull}1993[1993]
 
 *Metricbeat*
 
@@ -53,6 +54,7 @@ https://github.com/elastic/beats/compare/v5.0.0-alpha4...master[Check the HEAD d
 - Command line flag -c can be used multiple times. {pull}1985[1985]
 - Add OR/AND/NOT to the condition associated with the processors. {pull}1983[1983]
 - Add '-E' CLI flag for overwriting single config options via command line. {pull}1986[1986]
+- Choose the mapping template file based on the Elasticsearch version. {pull}1993[1993]
 
 *Metricbeat*
 
diff --git a/filebeat/filebeat.full.yml b/filebeat/filebeat.full.yml
index bdefa3dccb0..23d8e13b160 100644
--- a/filebeat/filebeat.full.yml
+++ b/filebeat/filebeat.full.yml
@@ -357,16 +357,28 @@ output.elasticsearch:
 
   # A template is used to set the mapping in Elasticsearch
   # By default template loading is enabled and the template is loaded.
-  # These settings can be adjusted to load your own template or overwrite existing ones
+  # These settings can be adjusted to load your own template or overwrite existing ones.
+
+  # Set to false to disable template loading.
+  #template.enabled: true
 
   # Template name. By default the template name is filebeat.
-  template.name: "filebeat"
+  #template.name: "filebeat"
 
   # Path to template file
-  template.path: "filebeat.template.json"
+  #template.path: "filebeat.template.json"
 
   # Overwrite existing template
-  template.overwrite: false
+  #template.overwrite: false
+
+  # If set to true, filebeat checks the Elasticsearch version at connect time, and if it
+  # is 2.x, it loads the file specified by the template.versions.2x.path setting. The 
+  # default is true.
+  #template.versions.2x.enabled: true
+
+  # Path to the Elasticsearch 2.x version of the template file.
+  #template.versions.2x.path: "filebeat.template-es2x.json"
+
 
   # TLS configuration. By default is off.
   # List of root certificates for HTTPS server verifications
diff --git a/filebeat/filebeat.yml b/filebeat/filebeat.yml
index 6f82628e096..ae30266d53c 100644
--- a/filebeat/filebeat.yml
+++ b/filebeat/filebeat.yml
@@ -82,15 +82,6 @@ output.elasticsearch:
   # Array of hosts to connect to.
   hosts: ["localhost:9200"]
 
-  # Template name. By default the template name is filebeat.
-  template.name: "filebeat"
-
-  # Path to template file
-  template.path: "filebeat.template.json"
-
-  # Overwrite existing template
-  template.overwrite: false
-
 #----------------------------- Logstash output --------------------------------
 #output.logstash:
   # The Logstash hosts
diff --git a/libbeat/_meta/config.full.yml b/libbeat/_meta/config.full.yml
index e81db63028f..d23ec609937 100644
--- a/libbeat/_meta/config.full.yml
+++ b/libbeat/_meta/config.full.yml
@@ -139,16 +139,28 @@ output.elasticsearch:
 
   # A template is used to set the mapping in Elasticsearch
   # By default template loading is enabled and the template is loaded.
-  # These settings can be adjusted to load your own template or overwrite existing ones
+  # These settings can be adjusted to load your own template or overwrite existing ones.
+
+  # Set to false to disable template loading.
+  #template.enabled: true
 
   # Template name. By default the template name is beatname.
-  template.name: "beatname"
+  #template.name: "beatname"
 
   # Path to template file
-  template.path: "beatname.template.json"
+  #template.path: "beatname.template.json"
 
   # Overwrite existing template
-  template.overwrite: false
+  #template.overwrite: false
+
+  # If set to true, beatname checks the Elasticsearch version at connect time, and if it
+  # is 2.x, it loads the file specified by the template.versions.2x.path setting. The 
+  # default is true.
+  #template.versions.2x.enabled: true
+
+  # Path to the Elasticsearch 2.x version of the template file.
+  #template.versions.2x.path: "beatname.template-es2x.json"
+
 
   # TLS configuration. By default is off.
   # List of root certificates for HTTPS server verifications
diff --git a/libbeat/_meta/config.yml b/libbeat/_meta/config.yml
index 283b56cf804..2e6cdfe46e1 100644
--- a/libbeat/_meta/config.yml
+++ b/libbeat/_meta/config.yml
@@ -24,15 +24,6 @@ output.elasticsearch:
   # Array of hosts to connect to.
   hosts: ["localhost:9200"]
 
-  # Template name. By default the template name is beatname.
-  template.name: "beatname"
-
-  # Path to template file
-  template.path: "beatname.template.json"
-
-  # Overwrite existing template
-  template.overwrite: false
-
 #----------------------------- Logstash output --------------------------------
 #output.logstash:
   # The Logstash hosts
diff --git a/libbeat/docs/outputconfig.asciidoc b/libbeat/docs/outputconfig.asciidoc
index 02bb6825251..b8abd7cb10b 100644
--- a/libbeat/docs/outputconfig.asciidoc
+++ b/libbeat/docs/outputconfig.asciidoc
@@ -24,8 +24,8 @@ output.elasticsearch:
   # The Elasticsearch cluster
   hosts: ["http://localhost:9200"]
 
-  # Template name. By default the template name is {beatname_lc}.
-  template.name: "{beatname_lc}"
+  # Load the Elasticsearch template.
+  template.enabled: true
 
   # Path to template file
   template.path: "{beatname_lc}.template.json"
@@ -195,6 +195,9 @@ enabled.
 
 You can adjust the following settings to load your own template or overwrite an existing one:
 
+*`enabled`*:: Set to false to disable template loading. If set this to false,
+you must <<load-template-manually,load the template manually>>.
+
 *`name`*:: The name of the template. The default is +{beatname_lc}+.
 
 *`path`*:: The path to the template file. The default is +{beatname_lc}.template.json+. If a relative
@@ -215,8 +218,28 @@ output.elasticsearch:
   template.overwrite: false
 ----------------------------------------------------------------------
 
-To disable automatic template loading, comment out the template part under elasticsearch output.
-If you disable this option, you must <<load-template-manually,load the template manually>>. 
+===== template.versions
+
+In the default configuration, {beatname_uc} automatically checks the
+Elasticsearch version and loads the recommended template file for the particular
+version. This behaviour can be controlled from the following options:
+
+*`2x.path`*:: The path to the template file to load for
+Elasticsearch versions 2.x.y. The default is +{beatname_lc}.template-es2x.json+.
+
+*`2x.enabled`*:: If set to +false+, the +2x.path+ option is ignored and the
+default template is loaded regardless of the Elasticsearch version.
+
+For example:
+
+["source","yaml",subs="attributes,callouts"]
+----------------------------------------------------------------------
+output.elasticsearch:
+  hosts: ["localhost:9200"]
+  template.path: "{beatname_lc}.template.json"
+  template.overwrite: false
+  template.versions.2x.path: "{beatname_lc}.template-es2x.json
+----------------------------------------------------------------------
 
 ===== max_retries
 
diff --git a/libbeat/outputs/console/console.go b/libbeat/outputs/console/console.go
index 520c8d41c83..dc15121f8de 100644
--- a/libbeat/outputs/console/console.go
+++ b/libbeat/outputs/console/console.go
@@ -18,7 +18,7 @@ type console struct {
 	config config
 }
 
-func New(config *common.Config, _ int) (outputs.Outputer, error) {
+func New(_ string, config *common.Config, _ int) (outputs.Outputer, error) {
 	c := &console{config: defaultConfig}
 	err := config.Unpack(&c.config)
 	if err != nil {
diff --git a/libbeat/outputs/elasticsearch/api_mock_test.go b/libbeat/outputs/elasticsearch/api_mock_test.go
index 0083d1a6c71..3578eea3f01 100644
--- a/libbeat/outputs/elasticsearch/api_mock_test.go
+++ b/libbeat/outputs/elasticsearch/api_mock_test.go
@@ -19,12 +19,14 @@ import (
 func ElasticsearchMock(code int, body []byte) *httptest.Server {
 
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		respCode := code
-		if r.Method == "HEAD" { // send ok on ping
-			respCode = 200
+		if r.URL.Path == "/" { // send ok and a minimal JSON on ping
+			w.WriteHeader(200)
+			w.Header().Set("Content-Type", "application/json")
+			w.Write([]byte(`{"version":{"number":"5.0.0"}}`))
+			return
 		}
 
-		w.WriteHeader(respCode)
+		w.WriteHeader(code)
 		if body != nil {
 			w.Header().Set("Content-Type", "application/json")
 			w.Write(body)
diff --git a/libbeat/outputs/elasticsearch/client.go b/libbeat/outputs/elasticsearch/client.go
index 6cf3cdc861c..044cc96d337 100644
--- a/libbeat/outputs/elasticsearch/client.go
+++ b/libbeat/outputs/elasticsearch/client.go
@@ -3,6 +3,7 @@ package elasticsearch
 import (
 	"bytes"
 	"crypto/tls"
+	"encoding/json"
 	"errors"
 	"expvar"
 	"fmt"
@@ -46,6 +47,7 @@ type Connection struct {
 	onConnectCallback func() error
 
 	encoder bodyEncoder
+	version string
 }
 
 // Metrics that can retrieved through the expvar web interface.
@@ -495,7 +497,7 @@ func (client *Client) CheckTemplate(templateName string) bool {
 
 func (conn *Connection) Connect(timeout time.Duration) error {
 	var err error
-	conn.connected, err = conn.Ping(timeout)
+	conn.connected, conn.version, err = conn.Ping(timeout)
 	if err != nil {
 		return err
 	}
@@ -510,18 +512,35 @@ func (conn *Connection) Connect(timeout time.Duration) error {
 	return nil
 }
 
-func (conn *Connection) Ping(timeout time.Duration) (bool, error) {
+// Ping sends a GET request to the Elasticsearch
+func (conn *Connection) Ping(timeout time.Duration) (bool, string, error) {
 	debugf("ES Ping(url=%v, timeout=%v)", conn.URL, timeout)
 
 	conn.http.Timeout = timeout
-	status, _, err := conn.execRequest("HEAD", conn.URL, nil)
+	status, body, err := conn.execRequest("GET", conn.URL, nil)
 	if err != nil {
 		debugf("Ping request failed with: %v", err)
-		return false, err
+		return false, "", err
+	}
+
+	if status >= 300 {
+		return false, "", fmt.Errorf("Non 2xx response code: %d", status)
+	}
+
+	var response struct {
+		Version struct {
+			Number string
+		}
+	}
+
+	err = json.Unmarshal(body, &response)
+	if err != nil {
+		return false, "", fmt.Errorf("Failed to parse JSON response: %v", err)
 	}
 
 	debugf("Ping status code: %v", status)
-	return status < 300, nil
+	logp.Info("Connected to Elasticsearch version %s", response.Version.Number)
+	return true, response.Version.Number, nil
 }
 
 func (conn *Connection) IsConnected() bool {
diff --git a/libbeat/outputs/elasticsearch/client_integration_test.go b/libbeat/outputs/elasticsearch/client_integration_test.go
index 5d09f1e1d3f..f040e3eb303 100644
--- a/libbeat/outputs/elasticsearch/client_integration_test.go
+++ b/libbeat/outputs/elasticsearch/client_integration_test.go
@@ -152,8 +152,9 @@ func TestOutputLoadTemplate(t *testing.T) {
 	config := map[string]interface{}{
 		"hosts": GetEsHost(),
 		"template": map[string]interface{}{
-			"name": "libbeat",
-			"path": tPath,
+			"name":                "libbeat",
+			"path":                tPath,
+			"versions.2x.enabled": false,
 		},
 	}
 
@@ -162,7 +163,7 @@ func TestOutputLoadTemplate(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	output, err := New(cfg, 0)
+	output, err := New("libbeat", cfg, 0)
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/libbeat/outputs/elasticsearch/config.go b/libbeat/outputs/elasticsearch/config.go
index e512e346c3a..7cf18de3fcc 100644
--- a/libbeat/outputs/elasticsearch/config.go
+++ b/libbeat/outputs/elasticsearch/config.go
@@ -24,9 +24,20 @@ type elasticsearchConfig struct {
 }
 
 type Template struct {
-	Name      string `config:"name"`
-	Path      string `config:"path"`
-	Overwrite bool   `config:"overwrite"`
+	Enabled   bool             `config:"enabled"`
+	Name      string           `config:"name"`
+	Path      string           `config:"path"`
+	Overwrite bool             `config:"overwrite"`
+	Versions  TemplateVersions `config:"versions"`
+}
+
+type TemplateVersions struct {
+	Es2x TemplateVersion `config:"2x"`
+}
+
+type TemplateVersion struct {
+	Enabled bool   `config:"enabled"`
+	Path    string `config:"path"`
 }
 
 const (
@@ -46,6 +57,10 @@ var (
 		CompressionLevel: 0,
 		TLS:              nil,
 		LoadBalance:      true,
+		Template: Template{
+			Enabled:  true,
+			Versions: TemplateVersions{Es2x: TemplateVersion{Enabled: true}},
+		},
 	}
 )
 
diff --git a/libbeat/outputs/elasticsearch/output.go b/libbeat/outputs/elasticsearch/output.go
index bda69964252..568fa8c1d69 100644
--- a/libbeat/outputs/elasticsearch/output.go
+++ b/libbeat/outputs/elasticsearch/output.go
@@ -21,11 +21,13 @@ import (
 )
 
 type elasticsearchOutput struct {
-	index string
-	mode  mode.ConnectionMode
+	index    string
+	beatName string
+	mode     mode.ConnectionMode
 	topology
 
 	template      map[string]interface{}
+	template2x    map[string]interface{}
 	templateMutex sync.Mutex
 }
 
@@ -49,12 +51,12 @@ var (
 )
 
 // New instantiates a new output plugin instance publishing to elasticsearch.
-func New(cfg *common.Config, topologyExpire int) (outputs.Outputer, error) {
+func New(beatName string, cfg *common.Config, topologyExpire int) (outputs.Outputer, error) {
 	if !cfg.HasField("bulk_max_size") {
 		cfg.SetInt("bulk_max_size", -1, defaultBulkSize)
 	}
 
-	output := &elasticsearchOutput{}
+	output := &elasticsearchOutput{beatName: beatName}
 	err := output.init(cfg, topologyExpire)
 	if err != nil {
 		return nil, err
@@ -76,7 +78,7 @@ func (out *elasticsearchOutput) init(
 		return err
 	}
 
-	err = out.readTemplate(config.Template)
+	err = out.readTemplate(&config.Template)
 	if err != nil {
 		return err
 	}
@@ -110,19 +112,40 @@ func (out *elasticsearchOutput) init(
 }
 
 // readTemplates reads the ES mapping template from the disk, if configured.
-func (out *elasticsearchOutput) readTemplate(config Template) error {
-	if len(config.Name) > 0 {
+func (out *elasticsearchOutput) readTemplate(config *Template) error {
+	if config.Enabled {
+		// Set the defaults that depend on the beat name
+		if config.Name == "" {
+			config.Name = out.beatName
+		}
+		if config.Path == "" {
+			config.Path = fmt.Sprintf("%s.template.json", out.beatName)
+		}
+		if config.Versions.Es2x.Path == "" {
+			config.Versions.Es2x.Path = fmt.Sprintf("%s.template-es2x.json", out.beatName)
+		}
+
 		// Look for the template in the configuration path, if it's not absolute
 		templatePath := paths.Resolve(paths.Config, config.Path)
-
 		logp.Info("Loading template enabled. Reading template file: %v", templatePath)
 
 		template, err := readTemplate(templatePath)
 		if err != nil {
 			return fmt.Errorf("Error loading template %s: %v", templatePath, err)
 		}
-
 		out.template = template
+
+		if config.Versions.Es2x.Enabled {
+			// Read the version of the template compatible with ES 2.x
+			templatePath := paths.Resolve(paths.Config, config.Versions.Es2x.Path)
+			logp.Info("Loading template enabled for Elasticsearch 2.x. Reading template file: %v", templatePath)
+
+			template, err := readTemplate(templatePath)
+			if err != nil {
+				return fmt.Errorf("Error loading template %s: %v", templatePath, err)
+			}
+			out.template2x = template
+		}
 	}
 	return nil
 }
@@ -161,7 +184,13 @@ func (out *elasticsearchOutput) loadTemplate(config Template, client *Client) er
 			logp.Info("Existing template will be overwritten, as overwrite is enabled.")
 		}
 
-		err := client.LoadTemplate(config.Name, out.template)
+		template := out.template
+		if config.Versions.Es2x.Enabled && strings.HasPrefix(client.Connection.version, "2.") {
+			logp.Info("Detected Elasticsearch 2.x. Automatically selecting the 2.x version of the template")
+			template = out.template2x
+		}
+
+		err := client.LoadTemplate(config.Name, template)
 		if err != nil {
 			return fmt.Errorf("Could not load template: %v", err)
 		}
diff --git a/libbeat/outputs/elasticsearch/output_test.go b/libbeat/outputs/elasticsearch/output_test.go
index 16155ef85ff..3eb32418b68 100644
--- a/libbeat/outputs/elasticsearch/output_test.go
+++ b/libbeat/outputs/elasticsearch/output_test.go
@@ -26,16 +26,17 @@ func createElasticsearchConnection(flushInterval int, bulkSize int) elasticsearc
 	}
 
 	config, _ := common.NewConfigFrom(map[string]interface{}{
-		"save_topology":  true,
-		"hosts":          []string{GetEsHost()},
-		"port":           esPort,
-		"username":       os.Getenv("ES_USER"),
-		"password":       os.Getenv("ES_PASS"),
-		"path":           "",
-		"index":          index,
-		"protocol":       "http",
-		"flush_interval": flushInterval,
-		"bulk_max_size":  bulkSize,
+		"save_topology":    true,
+		"hosts":            []string{GetEsHost()},
+		"port":             esPort,
+		"username":         os.Getenv("ES_USER"),
+		"password":         os.Getenv("ES_PASS"),
+		"path":             "",
+		"index":            index,
+		"protocol":         "http",
+		"flush_interval":   flushInterval,
+		"bulk_max_size":    bulkSize,
+		"template.enabled": false,
 	})
 
 	var output elasticsearchOutput
@@ -46,7 +47,7 @@ func createElasticsearchConnection(flushInterval int, bulkSize int) elasticsearc
 func TestTopologyInES(t *testing.T) {
 
 	if testing.Verbose() {
-		logp.LogInit(logp.LOG_DEBUG, "", false, true, []string{"topology", "output_elasticsearch"})
+		logp.LogInit(logp.LOG_DEBUG, "", false, true, []string{"*"})
 	}
 
 	elasticsearchOutput1 := createElasticsearchConnection(0, 0)
diff --git a/libbeat/outputs/fileout/file.go b/libbeat/outputs/fileout/file.go
index fb54049582c..c20558552c7 100644
--- a/libbeat/outputs/fileout/file.go
+++ b/libbeat/outputs/fileout/file.go
@@ -14,11 +14,12 @@ func init() {
 }
 
 type fileOutput struct {
-	rotator logp.FileRotator
+	beatName string
+	rotator  logp.FileRotator
 }
 
 // New instantiates a new file output instance.
-func New(cfg *common.Config, _ int) (outputs.Outputer, error) {
+func New(beatName string, cfg *common.Config, _ int) (outputs.Outputer, error) {
 	config := defaultConfig
 	if err := cfg.Unpack(&config); err != nil {
 		return nil, err
@@ -28,7 +29,7 @@ func New(cfg *common.Config, _ int) (outputs.Outputer, error) {
 	cfg.SetInt("flush_interval", -1, -1)
 	cfg.SetInt("bulk_max_size", -1, -1)
 
-	output := &fileOutput{}
+	output := &fileOutput{beatName: beatName}
 	if err := output.init(config); err != nil {
 		return nil, err
 	}
@@ -39,7 +40,7 @@ func (out *fileOutput) init(config config) error {
 	out.rotator.Path = config.Path
 	out.rotator.Name = config.Filename
 	if out.rotator.Name == "" {
-		out.rotator.Name = config.Index
+		out.rotator.Name = out.beatName
 	}
 	logp.Info("File output path set to: %v", out.rotator.Path)
 	logp.Info("File output base filename set to: %v", out.rotator.Name)
diff --git a/libbeat/outputs/kafka/kafka.go b/libbeat/outputs/kafka/kafka.go
index 42bbcbb0ced..ee01c98f66e 100644
--- a/libbeat/outputs/kafka/kafka.go
+++ b/libbeat/outputs/kafka/kafka.go
@@ -54,7 +54,7 @@ var (
 )
 
 // New instantiates a new kafka output instance.
-func New(cfg *common.Config, topologyExpire int) (outputs.Outputer, error) {
+func New(beatName string, cfg *common.Config, topologyExpire int) (outputs.Outputer, error) {
 	output := &kafka{}
 	err := output.init(cfg)
 	if err != nil {
diff --git a/libbeat/outputs/kafka/kafka_integration_test.go b/libbeat/outputs/kafka/kafka_integration_test.go
index b5624012736..f4689a5891e 100644
--- a/libbeat/outputs/kafka/kafka_integration_test.go
+++ b/libbeat/outputs/kafka/kafka_integration_test.go
@@ -69,7 +69,7 @@ func newTestKafkaOutput(t *testing.T, topic string, useType bool) outputs.Output
 		t.Fatal(err)
 	}
 
-	output, err := New(cfg, 0)
+	output, err := New("libbeat", cfg, 0)
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/libbeat/outputs/logstash/logstash.go b/libbeat/outputs/logstash/logstash.go
index edd19b91f85..b3197e0ea6e 100644
--- a/libbeat/outputs/logstash/logstash.go
+++ b/libbeat/outputs/logstash/logstash.go
@@ -46,7 +46,7 @@ func init() {
 	outputs.RegisterOutputPlugin("logstash", new)
 }
 
-func new(cfg *common.Config, _ int) (outputs.Outputer, error) {
+func new(beatName string, cfg *common.Config, _ int) (outputs.Outputer, error) {
 	output := &logstash{}
 	if err := output.init(cfg); err != nil {
 		return nil, err
diff --git a/libbeat/outputs/logstash/logstash_integration_test.go b/libbeat/outputs/logstash/logstash_integration_test.go
index 3ee2309be92..30825d39549 100644
--- a/libbeat/outputs/logstash/logstash_integration_test.go
+++ b/libbeat/outputs/logstash/logstash_integration_test.go
@@ -141,15 +141,16 @@ func newTestElasticsearchOutput(t *testing.T, test string) *testOutputer {
 	flushInterval := 0
 	bulkSize := 0
 	config, _ := common.NewConfigFrom(map[string]interface{}{
-		"hosts":          []string{getElasticsearchHost()},
-		"index":          index,
-		"flush_interval": &flushInterval,
-		"bulk_max_size":  &bulkSize,
-		"username":       os.Getenv("ES_USER"),
-		"password":       os.Getenv("ES_PASS"),
+		"hosts":            []string{getElasticsearchHost()},
+		"index":            index,
+		"flush_interval":   &flushInterval,
+		"bulk_max_size":    &bulkSize,
+		"username":         os.Getenv("ES_USER"),
+		"password":         os.Getenv("ES_PASS"),
+		"template.enabled": false,
 	})
 
-	output, err := plugin(config, 10)
+	output, err := plugin("libbeat", config, 10)
 	if err != nil {
 		t.Fatalf("init elasticsearch output plugin failed: %v", err)
 	}
diff --git a/libbeat/outputs/logstash/logstash_test.go b/libbeat/outputs/logstash/logstash_test.go
index d9a0ff1c26c..ad5a5ed4d89 100644
--- a/libbeat/outputs/logstash/logstash_test.go
+++ b/libbeat/outputs/logstash/logstash_test.go
@@ -74,7 +74,7 @@ func newTestLumberjackOutput(
 	}
 
 	cfg, _ := common.NewConfigFrom(config)
-	output, err := plugin(cfg, 0)
+	output, err := plugin("", cfg, 0)
 	if err != nil {
 		t.Fatalf("init logstash output plugin failed: %v", err)
 	}
diff --git a/libbeat/outputs/outputs.go b/libbeat/outputs/outputs.go
index 58e23f5d329..851e7472dd0 100644
--- a/libbeat/outputs/outputs.go
+++ b/libbeat/outputs/outputs.go
@@ -33,7 +33,7 @@ type BulkOutputer interface {
 }
 
 // Create and initialize the output plugin
-type OutputBuilder func(config *common.Config, topologyExpire int) (Outputer, error)
+type OutputBuilder func(beatName string, config *common.Config, topologyExpire int) (Outputer, error)
 
 // Functions to be exported by a output plugin
 type OutputInterface interface {
@@ -80,7 +80,7 @@ func InitOutputs(
 			config.SetString("index", -1, beatName)
 		}
 
-		output, err := plugin(config, topologyExpire)
+		output, err := plugin(beatName, config, topologyExpire)
 		if err != nil {
 			logp.Err("failed to initialize %s plugin as output: %s", name, err)
 			return nil, err
diff --git a/libbeat/outputs/redis/redis.go b/libbeat/outputs/redis/redis.go
index b0e6eeed194..e6fc53cf2c8 100644
--- a/libbeat/outputs/redis/redis.go
+++ b/libbeat/outputs/redis/redis.go
@@ -39,7 +39,7 @@ func init() {
 	outputs.RegisterOutputPlugin("redis", new)
 }
 
-func new(cfg *common.Config, expireTopo int) (outputs.Outputer, error) {
+func new(beatName string, cfg *common.Config, expireTopo int) (outputs.Outputer, error) {
 	r := &redisOut{}
 	if err := r.init(cfg, expireTopo); err != nil {
 		return nil, err
diff --git a/libbeat/outputs/redis/redis_integration_test.go b/libbeat/outputs/redis/redis_integration_test.go
index 226f24f4f73..fd20dae3b62 100644
--- a/libbeat/outputs/redis/redis_integration_test.go
+++ b/libbeat/outputs/redis/redis_integration_test.go
@@ -326,7 +326,7 @@ func newRedisTestingOutput(t *testing.T, cfg map[string]interface{}) *redisOut {
 		t.Fatalf("Failed to unpack topology_expire: %v", err)
 	}
 
-	out, err := plugin(config, params.Expire)
+	out, err := plugin("libbeat", config, params.Expire)
 	if err != nil {
 		t.Fatalf("Failed to initialize redis output: %v", err)
 	}
diff --git a/libbeat/tests/system/test_base.py b/libbeat/tests/system/test_base.py
index c7c6ed8678a..a699b29caea 100644
--- a/libbeat/tests/system/test_base.py
+++ b/libbeat/tests/system/test_base.py
@@ -68,7 +68,9 @@ def test_config_test(self):
         """
         shutil.copy("../../_meta/config.yml",
                     os.path.join(self.working_dir, "libbeat.yml"))
-        with open(self.working_dir + "/beatname.template.json", "w") as f:
+        with open(self.working_dir + "/mockbeat.template.json", "w") as f:
+            f.write('{"template": true}')
+        with open(self.working_dir + "/mockbeat.template-es2x.json", "w") as f:
             f.write('{"template": true}')
 
         exit_code = self.run_beat(
diff --git a/metricbeat/metricbeat.full.yml b/metricbeat/metricbeat.full.yml
index dca2c98d1a6..92ae0cd3cef 100644
--- a/metricbeat/metricbeat.full.yml
+++ b/metricbeat/metricbeat.full.yml
@@ -284,16 +284,28 @@ output.elasticsearch:
 
   # A template is used to set the mapping in Elasticsearch
   # By default template loading is enabled and the template is loaded.
-  # These settings can be adjusted to load your own template or overwrite existing ones
+  # These settings can be adjusted to load your own template or overwrite existing ones.
+
+  # Set to false to disable template loading.
+  #template.enabled: true
 
   # Template name. By default the template name is metricbeat.
-  template.name: "metricbeat"
+  #template.name: "metricbeat"
 
   # Path to template file
-  template.path: "metricbeat.template.json"
+  #template.path: "metricbeat.template.json"
 
   # Overwrite existing template
-  template.overwrite: false
+  #template.overwrite: false
+
+  # If set to true, metricbeat checks the Elasticsearch version at connect time, and if it
+  # is 2.x, it loads the file specified by the template.versions.2x.path setting. The 
+  # default is true.
+  #template.versions.2x.enabled: true
+
+  # Path to the Elasticsearch 2.x version of the template file.
+  #template.versions.2x.path: "metricbeat.template-es2x.json"
+
 
   # TLS configuration. By default is off.
   # List of root certificates for HTTPS server verifications
diff --git a/metricbeat/metricbeat.yml b/metricbeat/metricbeat.yml
index ac2d63dad4a..abf95938b34 100644
--- a/metricbeat/metricbeat.yml
+++ b/metricbeat/metricbeat.yml
@@ -70,15 +70,6 @@ output.elasticsearch:
   # Array of hosts to connect to.
   hosts: ["localhost:9200"]
 
-  # Template name. By default the template name is metricbeat.
-  template.name: "metricbeat"
-
-  # Path to template file
-  template.path: "metricbeat.template.json"
-
-  # Overwrite existing template
-  template.overwrite: false
-
 #----------------------------- Logstash output --------------------------------
 #output.logstash:
   # The Logstash hosts
diff --git a/packetbeat/packetbeat.full.yml b/packetbeat/packetbeat.full.yml
index 859fcf2867d..bc39ec60991 100644
--- a/packetbeat/packetbeat.full.yml
+++ b/packetbeat/packetbeat.full.yml
@@ -523,16 +523,28 @@ output.elasticsearch:
 
   # A template is used to set the mapping in Elasticsearch
   # By default template loading is enabled and the template is loaded.
-  # These settings can be adjusted to load your own template or overwrite existing ones
+  # These settings can be adjusted to load your own template or overwrite existing ones.
+
+  # Set to false to disable template loading.
+  #template.enabled: true
 
   # Template name. By default the template name is packetbeat.
-  template.name: "packetbeat"
+  #template.name: "packetbeat"
 
   # Path to template file
-  template.path: "packetbeat.template.json"
+  #template.path: "packetbeat.template.json"
 
   # Overwrite existing template
-  template.overwrite: false
+  #template.overwrite: false
+
+  # If set to true, packetbeat checks the Elasticsearch version at connect time, and if it
+  # is 2.x, it loads the file specified by the template.versions.2x.path setting. The 
+  # default is true.
+  #template.versions.2x.enabled: true
+
+  # Path to the Elasticsearch 2.x version of the template file.
+  #template.versions.2x.path: "packetbeat.template-es2x.json"
+
 
   # TLS configuration. By default is off.
   # List of root certificates for HTTPS server verifications
diff --git a/packetbeat/packetbeat.yml b/packetbeat/packetbeat.yml
index 7bc20d44096..cce91765bec 100644
--- a/packetbeat/packetbeat.yml
+++ b/packetbeat/packetbeat.yml
@@ -113,15 +113,6 @@ output.elasticsearch:
   # Array of hosts to connect to.
   hosts: ["localhost:9200"]
 
-  # Template name. By default the template name is packetbeat.
-  template.name: "packetbeat"
-
-  # Path to template file
-  template.path: "packetbeat.template.json"
-
-  # Overwrite existing template
-  template.overwrite: false
-
 #----------------------------- Logstash output --------------------------------
 #output.logstash:
   # The Logstash hosts
diff --git a/winlogbeat/winlogbeat.full.yml b/winlogbeat/winlogbeat.full.yml
index 97b47c43df6..651434ed0a4 100644
--- a/winlogbeat/winlogbeat.full.yml
+++ b/winlogbeat/winlogbeat.full.yml
@@ -174,16 +174,28 @@ output.elasticsearch:
 
   # A template is used to set the mapping in Elasticsearch
   # By default template loading is enabled and the template is loaded.
-  # These settings can be adjusted to load your own template or overwrite existing ones
+  # These settings can be adjusted to load your own template or overwrite existing ones.
+
+  # Set to false to disable template loading.
+  #template.enabled: true
 
   # Template name. By default the template name is winlogbeat.
-  template.name: "winlogbeat"
+  #template.name: "winlogbeat"
 
   # Path to template file
-  template.path: "winlogbeat.template.json"
+  #template.path: "winlogbeat.template.json"
 
   # Overwrite existing template
-  template.overwrite: false
+  #template.overwrite: false
+
+  # If set to true, winlogbeat checks the Elasticsearch version at connect time, and if it
+  # is 2.x, it loads the file specified by the template.versions.2x.path setting. The 
+  # default is true.
+  #template.versions.2x.enabled: true
+
+  # Path to the Elasticsearch 2.x version of the template file.
+  #template.versions.2x.path: "winlogbeat.template-es2x.json"
+
 
   # TLS configuration. By default is off.
   # List of root certificates for HTTPS server verifications
diff --git a/winlogbeat/winlogbeat.yml b/winlogbeat/winlogbeat.yml
index c6428c9950f..5ebf7c42935 100644
--- a/winlogbeat/winlogbeat.yml
+++ b/winlogbeat/winlogbeat.yml
@@ -48,15 +48,6 @@ output.elasticsearch:
   # Array of hosts to connect to.
   hosts: ["localhost:9200"]
 
-  # Template name. By default the template name is winlogbeat.
-  template.name: "winlogbeat"
-
-  # Path to template file
-  template.path: "winlogbeat.template.json"
-
-  # Overwrite existing template
-  template.overwrite: false
-
 #----------------------------- Logstash output --------------------------------
 #output.logstash:
   # The Logstash hosts