Introduce beat version in index and template

This adds the beat version to the index and the template. The advantage of this is that an index always has the correct template applied based on the beats version. Currently during upgrade it must be waited until the next day. Only then the new template is applied. Also it helps in case different versions of beats are running in parallel. Each beat puts its data into the correct index. The index name is now `beatname-%{[beat.version]}-%{+yyyy.MM.dd}` by default. This should not have any affects on the dashboards or Kibana in general as `beatname-*` still applies to all data. Are there some potential side affects of this?
elastic · Feb 20, 2017 · 32fcdce · 32fcdce
1 parent 2eb2571
commit 32fcdce
Show file tree

Hide file tree

Showing 31 changed files with 66 additions and 49 deletions.
diff --git a/filebeat/filebeat.full.yml b/filebeat/filebeat.full.yml
@@ -454,7 +454,7 @@ output.elasticsearch:
 
   # Optional index name. The default is "filebeat" plus date
   # and generates [filebeat-]YYYY.MM.DD keys.
-  #index: "filebeat-%{+yyyy.MM.dd}"
+  #index: "filebeat-%{[beat.version]}-%{+yyyy.MM.dd}"
 
   # Optional ingest node pipeline. By default no pipeline will be used.
   #pipeline: ""
@@ -494,6 +494,8 @@ output.elasticsearch:
   #template.enabled: true
 
   # Template name. By default the template name is filebeat.
+  # The version of the beat will always be appended to the given name
+  # so the final name is filebeat-%{[beat.version]}.
   #template.name: "filebeat"
 
   # Path to template file

diff --git a/filebeat/filebeat.template-es2x.json b/filebeat/filebeat.template-es2x.json
@@ -537,5 +537,5 @@
   "settings": {
     "index.refresh_interval": "5s"
   },
-  "template": "filebeat-*"
+  "template": "filebeat-6.0.0-alpha1-*"
 }
diff --git a/filebeat/filebeat.template.json b/filebeat/filebeat.template.json
@@ -458,5 +458,5 @@
     "index.mapping.total_fields.limit": 10000,
     "index.refresh_interval": "5s"
   },
-  "template": "filebeat-*"
+  "template": "filebeat-6.0.0-alpha1-*"
 }
diff --git a/heartbeat/heartbeat.full.yml b/heartbeat/heartbeat.full.yml
@@ -302,7 +302,7 @@ output.elasticsearch:
 
   # Optional index name. The default is "heartbeat" plus date
   # and generates [heartbeat-]YYYY.MM.DD keys.
-  #index: "heartbeat-%{+yyyy.MM.dd}"
+  #index: "heartbeat-%{[beat.version]}-%{+yyyy.MM.dd}"
 
   # Optional ingest node pipeline. By default no pipeline will be used.
   #pipeline: ""
@@ -342,6 +342,8 @@ output.elasticsearch:
   #template.enabled: true
 
   # Template name. By default the template name is heartbeat.
+  # The version of the beat will always be appended to the given name
+  # so the final name is heartbeat-%{[beat.version]}.
   #template.name: "heartbeat"
 
   # Path to template file

diff --git a/heartbeat/heartbeat.template-es2x.json b/heartbeat/heartbeat.template-es2x.json
@@ -215,5 +215,5 @@
   "settings": {
     "index.refresh_interval": "5s"
   },
-  "template": "heartbeat-*"
+  "template": "heartbeat-6.0.0-alpha1-*"
 }
diff --git a/heartbeat/heartbeat.template.json b/heartbeat/heartbeat.template.json
@@ -188,5 +188,5 @@
     "index.mapping.total_fields.limit": 10000,
     "index.refresh_interval": "5s"
   },
-  "template": "heartbeat-*"
+  "template": "heartbeat-6.0.0-alpha1-*"
 }
diff --git a/libbeat/_meta/config.full.yml b/libbeat/_meta/config.full.yml
@@ -104,7 +104,7 @@ output.elasticsearch:
 
   # Optional index name. The default is "beatname" plus date
   # and generates [beatname-]YYYY.MM.DD keys.
-  #index: "beatname-%{+yyyy.MM.dd}"
+  #index: "beatname-%{[beat.version]}-%{+yyyy.MM.dd}"
 
   # Optional ingest node pipeline. By default no pipeline will be used.
   #pipeline: ""
@@ -144,6 +144,8 @@ output.elasticsearch:
   #template.enabled: true
 
   # Template name. By default the template name is beatname.
+  # The version of the beat will always be appended to the given name
+  # so the final name is beatname-%{[beat.version]}.
   #template.name: "beatname"
 
   # Path to template file

diff --git a/libbeat/outputs/console/console.go b/libbeat/outputs/console/console.go
@@ -21,7 +21,7 @@ type console struct {
 	codec outputs.Codec
 }
 
-func New(_ string, config *common.Config, _ int) (outputs.Outputer, error) {
+func New(_ common.BeatInfo, config *common.Config, _ int) (outputs.Outputer, error) {
 	var unpackedConfig Config
 	err := config.Unpack(&unpackedConfig)
 	if err != nil {

diff --git a/libbeat/outputs/elasticsearch/client_integration_test.go b/libbeat/outputs/elasticsearch/client_integration_test.go
@@ -181,7 +181,7 @@ func TestOutputLoadTemplate(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	output, err := New("libbeat", cfg, 0)
+	output, err := New(common.BeatInfo{Beat: "libbeat"}, cfg, 0)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -190,6 +190,9 @@ func TestOutputLoadTemplate(t *testing.T) {
 		"host":       "test-host",
 		"type":       "libbeat",
 		"message":    "Test message from libbeat",
+		"beat": common.MapStr{
+			"version": "1.2.3",
+		},
 	}}
 
 	err = output.PublishEvent(nil, outputs.Options{Guaranteed: true}, event)
@@ -423,7 +426,7 @@ func connectTestEs(t *testing.T, cfg interface{}) (outputs.BulkOutputer, *Client
 		t.Fatal(err)
 	}
 
-	output, err := New("libbeat", config, 0)
+	output, err := New(common.BeatInfo{Beat: "libbeat"}, config, 0)
 	if err != nil {
 		t.Fatal(err)
 	}

diff --git a/libbeat/outputs/elasticsearch/output.go b/libbeat/outputs/elasticsearch/output.go
@@ -23,7 +23,7 @@ import (
 
 type elasticsearchOutput struct {
 	index    outil.Selector
-	beatName string
+	beat     common.BeatInfo
 	pipeline *outil.Selector
 
 	mode mode.ConnectionMode
@@ -54,17 +54,17 @@ var (
 )
 
 // New instantiates a new output plugin instance publishing to elasticsearch.
-func New(beatName string, cfg *common.Config, topologyExpire int) (outputs.Outputer, error) {
+func New(beat common.BeatInfo, cfg *common.Config, topologyExpire int) (outputs.Outputer, error) {
 	if !cfg.HasField("bulk_max_size") {
 		cfg.SetInt("bulk_max_size", -1, defaultBulkSize)
 	}
 
 	if !cfg.HasField("index") {
-		pattern := fmt.Sprintf("%v-%%{+yyyy.MM.dd}", beatName)
+		pattern := fmt.Sprintf("%v-%v-%%{+yyyy.MM.dd}", beat.Beat, beat.Version)
 		cfg.SetString("index", -1, pattern)
 	}
 
-	output := &elasticsearchOutput{beatName: beatName}
+	output := &elasticsearchOutput{beat: beat}
 	err := output.init(cfg, topologyExpire)
 	if err != nil {
 		return nil, err
@@ -238,13 +238,13 @@ 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
+			config.Name = out.beat.Beat + "-" + out.beat.Version
 		}
 		if config.Path == "" {
-			config.Path = fmt.Sprintf("%s.template.json", out.beatName)
+			config.Path = fmt.Sprintf("%s.template.json", out.beat.Beat)
 		}
 		if config.Versions.Es2x.Path == "" {
-			config.Versions.Es2x.Path = fmt.Sprintf("%s.template-es2x.json", out.beatName)
+			config.Versions.Es2x.Path = fmt.Sprintf("%s.template-es2x.json", out.beat.Beat)
 		}
 
 		// Look for the template in the configuration path, if it's not absolute

diff --git a/libbeat/outputs/elasticsearch/output_test.go b/libbeat/outputs/elasticsearch/output_test.go
@@ -39,7 +39,7 @@ func createElasticsearchConnection(flushInterval int, bulkSize int) *elasticsear
 		"template.enabled": false,
 	})
 
-	output := &elasticsearchOutput{beatName: "test"}
+	output := &elasticsearchOutput{beat: common.BeatInfo{Beat: "test"}}
 	output.init(config, 10)
 	return output
 }

diff --git a/libbeat/outputs/fileout/file.go b/libbeat/outputs/fileout/file.go
@@ -12,13 +12,13 @@ func init() {
 }
 
 type fileOutput struct {
-	beatName string
-	rotator  logp.FileRotator
-	codec    outputs.Codec
+	beat    common.BeatInfo
+	rotator logp.FileRotator
+	codec   outputs.Codec
 }
 
 // New instantiates a new file output instance.
-func New(beatName string, cfg *common.Config, _ int) (outputs.Outputer, error) {
+func New(beat common.BeatInfo, cfg *common.Config, _ int) (outputs.Outputer, error) {
 	config := defaultConfig
 	if err := cfg.Unpack(&config); err != nil {
 		return nil, err
@@ -28,7 +28,7 @@ func New(beatName string, cfg *common.Config, _ int) (outputs.Outputer, error) {
 	cfg.SetInt("flush_interval", -1, -1)
 	cfg.SetInt("bulk_max_size", -1, -1)
 
-	output := &fileOutput{beatName: beatName}
+	output := &fileOutput{beat: beat}
 	if err := output.init(config); err != nil {
 		return nil, err
 	}
@@ -41,7 +41,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 = out.beatName
+		out.rotator.Name = out.beat.Beat
 	}
 
 	codec, err := outputs.CreateEncoder(config.Codec)

diff --git a/libbeat/outputs/kafka/kafka.go b/libbeat/outputs/kafka/kafka.go
@@ -100,7 +100,7 @@ var (
 )
 
 // New instantiates a new kafka output instance.
-func New(beatName string, cfg *common.Config, topologyExpire int) (outputs.Outputer, error) {
+func New(_ common.BeatInfo, 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
@@ -196,7 +196,7 @@ func TestKafkaPublish(t *testing.T) {
 		// create output within function scope to guarantee
 		// output is properly closed between single tests
 		func() {
-			tmp, err := New("libbeat", cfg, 0)
+			tmp, err := New(common.BeatInfo{Beat: "libbeat"}, cfg, 0)
 			if err != nil {
 				t.Fatal(err)
 			}

diff --git a/libbeat/outputs/logstash/logstash.go b/libbeat/outputs/logstash/logstash.go
@@ -46,10 +46,10 @@ func init() {
 	outputs.RegisterOutputPlugin("logstash", new)
 }
 
-func new(beatName string, cfg *common.Config, _ int) (outputs.Outputer, error) {
+func new(beat common.BeatInfo, cfg *common.Config, _ int) (outputs.Outputer, error) {
 
 	if !cfg.HasField("index") {
-		cfg.SetString("index", -1, beatName)
+		cfg.SetString("index", -1, beat.Beat)
 	}
 
 	output := &logstash{}

diff --git a/libbeat/outputs/logstash/logstash_integration_test.go b/libbeat/outputs/logstash/logstash_integration_test.go
@@ -160,7 +160,7 @@ func newTestElasticsearchOutput(t *testing.T, test string) *testOutputer {
 		"template.enabled": false,
 	})
 
-	output, err := plugin("libbeat", config, 10)
+	output, err := plugin(common.BeatInfo{Beat: "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
@@ -74,7 +74,7 @@ func newTestLumberjackOutput(
 	}
 
 	cfg, _ := common.NewConfigFrom(config)
-	output, err := plugin("", cfg, 0)
+	output, err := plugin(common.BeatInfo{}, 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
@@ -53,7 +53,7 @@ type BulkOutputer interface {
 }
 
 // Create and initialize the output plugin
-type OutputBuilder func(beatName string, config *common.Config, topologyExpire int) (Outputer, error)
+type OutputBuilder func(beat common.BeatInfo, config *common.Config, topologyExpire int) (Outputer, error)
 
 // Functions to be exported by a output plugin
 type OutputInterface interface {
@@ -90,7 +90,7 @@ func FindOutputPlugin(name string) OutputBuilder {
 }
 
 func InitOutputs(
-	beatName string,
+	beat common.BeatInfo,
 	configs map[string]*common.Config,
 	topologyExpire int,
 ) ([]OutputPlugin, error) {
@@ -106,7 +106,7 @@ func InitOutputs(
 			continue
 		}
 
-		output, err := plugin(beatName, config, topologyExpire)
+		output, err := plugin(beat, 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
@@ -18,7 +18,7 @@ import (
 type redisOut struct {
 	mode mode.ConnectionMode
 	topology
-	beatName string
+	beat common.BeatInfo
 }
 
 var debugf = logp.MakeDebug("redis")
@@ -40,8 +40,8 @@ func init() {
 	outputs.RegisterOutputPlugin("redis", new)
 }
 
-func new(beatName string, cfg *common.Config, expireTopo int) (outputs.Outputer, error) {
-	r := &redisOut{beatName: beatName}
+func new(beat common.BeatInfo, cfg *common.Config, expireTopo int) (outputs.Outputer, error) {
+	r := &redisOut{beat: beat}
 	if err := r.init(cfg, expireTopo); err != nil {
 		return nil, err
 	}
@@ -80,7 +80,7 @@ func (r *redisOut) init(cfg *common.Config, expireTopo int) error {
 		}
 	}
 	if !cfg.HasField("key") {
-		cfg.SetString("key", -1, r.beatName)
+		cfg.SetString("key", -1, r.beat.Beat)
 	}
 
 	key, err := outil.BuildSelectorFromConfig(cfg, outil.Settings{

diff --git a/libbeat/outputs/redis/redis_integration_test.go b/libbeat/outputs/redis/redis_integration_test.go
@@ -354,7 +354,7 @@ func newRedisTestingOutput(t *testing.T, cfg map[string]interface{}) *redisOut {
 		t.Fatalf("Failed to unpack topology_expire: %v", err)
 	}
 
-	out, err := plugin("libbeat", config, params.Expire)
+	out, err := plugin(common.BeatInfo{Beat: "libbeat"}, config, params.Expire)
 	if err != nil {
 		t.Fatalf("Failed to initialize redis output: %v", err)
 	}

diff --git a/libbeat/publisher/publish.go b/libbeat/publisher/publish.go
@@ -221,7 +221,8 @@ func (publisher *BeatPublisher) init(
 	publisher.wsOutput.Init()
 
 	if !publisher.disabled {
-		plugins, err := outputs.InitOutputs(beat.Beat, configs, shipper.TopologyExpire)
+		plugins, err := outputs.InitOutputs(beat, configs, shipper.TopologyExpire)
+
 		if err != nil {
 			return err
 		}

diff --git a/libbeat/scripts/generate_template.py b/libbeat/scripts/generate_template.py
@@ -377,4 +377,5 @@ def fill_field_properties(args, field, defaults, path):
             version_data = yaml.load(file)
 
         with open(target, 'w') as output:
-            fields_to_es_template(args, fields, output, args.beatname + "-*", version_data['version'])
+            fields_to_es_template(args, fields, output, args.beatname + "-" +
+                                  version_data['version'] + "-*", version_data['version'])
diff --git a/metricbeat/metricbeat.full.yml b/metricbeat/metricbeat.full.yml
@@ -410,7 +410,7 @@ output.elasticsearch:
 
   # Optional index name. The default is "metricbeat" plus date
   # and generates [metricbeat-]YYYY.MM.DD keys.
-  #index: "metricbeat-%{+yyyy.MM.dd}"
+  #index: "metricbeat-%{[beat.version]}-%{+yyyy.MM.dd}"
 
   # Optional ingest node pipeline. By default no pipeline will be used.
   #pipeline: ""
@@ -450,6 +450,8 @@ output.elasticsearch:
   #template.enabled: true
 
   # Template name. By default the template name is metricbeat.
+  # The version of the beat will always be appended to the given name
+  # so the final name is metricbeat-%{[beat.version]}.
   #template.name: "metricbeat"
 
   # Path to template file

diff --git a/metricbeat/metricbeat.template-es2x.json b/metricbeat/metricbeat.template-es2x.json
@@ -4186,5 +4186,5 @@
   "settings": {
     "index.refresh_interval": "5s"
   },
-  "template": "metricbeat-*"
+  "template": "metricbeat-6.0.0-alpha1-*"
 }
diff --git a/metricbeat/metricbeat.template.json b/metricbeat/metricbeat.template.json
@@ -4128,5 +4128,5 @@
     "index.mapping.total_fields.limit": 10000,
     "index.refresh_interval": "5s"
   },
-  "template": "metricbeat-*"
+  "template": "metricbeat-6.0.0-alpha1-*"
 }
diff --git a/packetbeat/packetbeat.full.yml b/packetbeat/packetbeat.full.yml
@@ -558,7 +558,7 @@ output.elasticsearch:
 
   # Optional index name. The default is "packetbeat" plus date
   # and generates [packetbeat-]YYYY.MM.DD keys.
-  #index: "packetbeat-%{+yyyy.MM.dd}"
+  #index: "packetbeat-%{[beat.version]}-%{+yyyy.MM.dd}"
 
   # Optional ingest node pipeline. By default no pipeline will be used.
   #pipeline: ""
@@ -598,6 +598,8 @@ output.elasticsearch:
   #template.enabled: true
 
   # Template name. By default the template name is packetbeat.
+  # The version of the beat will always be appended to the given name
+  # so the final name is packetbeat-%{[beat.version]}.
   #template.name: "packetbeat"
 
   # Path to template file