Add ILM support to Beats (#7963)

This adds support to Beats for index lifecycle management. With `output.elasticsearch.ilm.enabled: true` ilm can be enabled. It will overwrite the current index setting, automatically set up an alias with a write index and starting ingesting data to this alias. A default ILM template is provided with `30d / 50GB` as rollover policy.

CHANGELOG.asciidoc

@@ -72,6 +72,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha1...master[Check the HEAD d
 - Unify dashboard exporter tools. {pull}9097[9097]
 - Use _doc as document type of the Elasticsearch major version is 7. {pull}9056[9056]
 - Add cache.ttl to add_host_metadata. {pull}9359[9359]
+- Add support for index lifecycle management (beta). {pull}7963[7963]
 
 *Auditbeat*
 

auditbeat/auditbeat.reference.yml

@@ -357,6 +357,11 @@ output.elasticsearch:
   # IPv6 addresses should always be defined as: https://[2001:db8::1]:9200
   hosts: ["localhost:9200"]
 
+  # Enabled ilm (beta) to use index lifecycle management instead daily indices.
+  #ilm.enabled: false
+  #ilm.rollover_alias: "auditbeat"
+  #ilm.pattern: "{now/d}-000001"
+
   # Set gzip compression level.
   #compression_level: 0
 

auditbeat/auditbeat.yml

@@ -122,6 +122,9 @@ output.elasticsearch:
   # Array of hosts to connect to.
   hosts: ["localhost:9200"]
 
+  # Enabled ilm (beta) to use index lifecycle management instead daily indices.
+  #ilm.enabled: false
+
   # Optional protocol and basic auth credentials.
   #protocol: "https"
   #username: "elastic"

filebeat/filebeat.reference.yml

@@ -1031,6 +1031,11 @@ output.elasticsearch:
   # IPv6 addresses should always be defined as: https://[2001:db8::1]:9200
   hosts: ["localhost:9200"]
 
+  # Enabled ilm (beta) to use index lifecycle management instead daily indices.
+  #ilm.enabled: false
+  #ilm.rollover_alias: "filebeat"
+  #ilm.pattern: "{now/d}-000001"
+
   # Set gzip compression level.
   #compression_level: 0
 

filebeat/filebeat.yml

@@ -149,6 +149,9 @@ output.elasticsearch:
   # Array of hosts to connect to.
   hosts: ["localhost:9200"]
 
+  # Enabled ilm (beta) to use index lifecycle management instead daily indices.
+  #ilm.enabled: false
+
   # Optional protocol and basic auth credentials.
   #protocol: "https"
   #username: "elastic"

heartbeat/heartbeat.reference.yml

@@ -490,6 +490,11 @@ output.elasticsearch:
   # IPv6 addresses should always be defined as: https://[2001:db8::1]:9200
   hosts: ["localhost:9200"]
 
+  # Enabled ilm (beta) to use index lifecycle management instead daily indices.
+  #ilm.enabled: false
+  #ilm.rollover_alias: "heartbeat"
+  #ilm.pattern: "{now/d}-000001"
+
   # Set gzip compression level.
   #compression_level: 0
 

heartbeat/heartbeat.yml

@@ -96,6 +96,9 @@ output.elasticsearch:
   # Array of hosts to connect to.
   hosts: ["localhost:9200"]
 
+  # Enabled ilm (beta) to use index lifecycle management instead daily indices.
+  #ilm.enabled: false
+
   # Optional protocol and basic auth credentials.
   #protocol: "https"
   #username: "elastic"

journalbeat/journalbeat.reference.yml

@@ -291,6 +291,11 @@ output.elasticsearch:
   # IPv6 addresses should always be defined as: https://[2001:db8::1]:9200
   hosts: ["localhost:9200"]
 
+  # Enabled ilm (beta) to use index lifecycle management instead daily indices.
+  #ilm.enabled: false
+  #ilm.rollover_alias: "journalbeat"
+  #ilm.pattern: "{now/d}-000001"
+
   # Set gzip compression level.
   #compression_level: 0
 

journalbeat/journalbeat.yml

@@ -112,6 +112,9 @@ output.elasticsearch:
   # Array of hosts to connect to.
   hosts: ["localhost:9200"]
 
+  # Enabled ilm (beta) to use index lifecycle management instead daily indices.
+  #ilm.enabled: false
+
   # Optional protocol and basic auth credentials.
   #protocol: "https"
   #username: "elastic"

libbeat/_meta/config.reference.yml

@@ -245,6 +245,11 @@ output.elasticsearch:
   # IPv6 addresses should always be defined as: https://[2001:db8::1]:9200
   hosts: ["localhost:9200"]
 
+  # Enabled ilm (beta) to use index lifecycle management instead daily indices.
+  #ilm.enabled: false
+  #ilm.rollover_alias: "beat-index-prefix"
+  #ilm.pattern: "{now/d}-000001"
+
   # Set gzip compression level.
   #compression_level: 0
 

libbeat/_meta/config.yml

@@ -66,6 +66,9 @@ output.elasticsearch:
   # Array of hosts to connect to.
   hosts: ["localhost:9200"]
 
+  # Enabled ilm (beta) to use index lifecycle management instead daily indices.
+  #ilm.enabled: false
+
   # Optional protocol and basic auth credentials.
   #protocol: "https"
   #username: "elastic"

libbeat/cmd/export.go

@@ -33,6 +33,7 @@ func genExportCmd(settings instance.Settings, name, idxPrefix, beatVersion strin
 	exportCmd.AddCommand(export.GenExportConfigCmd(settings, name, idxPrefix, beatVersion))
 	exportCmd.AddCommand(export.GenTemplateConfigCmd(settings, name, idxPrefix, beatVersion))
 	exportCmd.AddCommand(export.GenDashboardCmd(name, idxPrefix, beatVersion))
+	exportCmd.AddCommand(export.GenGetILMPolicyCmd())
 
 	return exportCmd
 }

libbeat/cmd/export/ilm_policy.go

@@ -0,0 +1,39 @@
+// 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 (
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	"github.com/elastic/beats/libbeat/cmd/instance"
+)
+
+// GenGetILMPolicyCmd is the command used to export the ilm policy.
+func GenGetILMPolicyCmd() *cobra.Command {
+	genTemplateConfigCmd := &cobra.Command{
+		Use:   "ilm-policy",
+		Short: "Export ILM policy",
+		Run: func(cmd *cobra.Command, args []string) {
+			fmt.Println(instance.ILMPolicy.StringToPrint())
+		},
+	}
+
+	return genTemplateConfigCmd
+}

libbeat/cmd/instance/beat.go

@@ -105,6 +105,9 @@ type beatConfig struct {
 	Dashboards *common.Config `config:"setup.dashboards"`
 	Template   *common.Config `config:"setup.template"`
 	Kibana     *common.Config `config:"setup.kibana"`
+
+	// ILM Config options
+	ILM *common.Config `config:"output.elasticsearch.ilm"`
 }
 
 var (
@@ -430,7 +433,7 @@ func (b *Beat) TestConfig(bt beat.Creator) error {
 }
 
 // Setup registers ES index template, kibana dashboards, ml jobs and pipelines.
-func (b *Beat) Setup(bt beat.Creator, template, setupDashboards, machineLearning, pipelines bool) error {
+func (b *Beat) Setup(bt beat.Creator, template, setupDashboards, machineLearning, pipelines, policy bool) error {
 	return handleError(func() error {
 		err := b.Init()
 		if err != nil {
@@ -509,6 +512,13 @@ func (b *Beat) Setup(bt beat.Creator, template, setupDashboards, machineLearning
 			fmt.Println("Loaded Ingest pipelines")
 		}
 
+		if policy {
+			if err := b.loadILMPolicy(); err != nil {
+				return err
+			}
+			fmt.Println("Loaded Index Lifecycle Management (ILM) policy")
+		}
+
 		return nil
 	}())
 }
@@ -719,11 +729,11 @@ func (b *Beat) loadDashboards(ctx context.Context, force bool) error {
 // the elasticsearch output. It is important the the registration happens before
 // the publisher is created.
 func (b *Beat) registerTemplateLoading() error {
-	var cfg template.TemplateConfig
+	var templateCfg template.TemplateConfig
 
 	// Check if outputting to file is enabled, and output to file if it is
 	if b.Config.Template.Enabled() {
-		err := b.Config.Template.Unpack(&cfg)
+		err := b.Config.Template.Unpack(&templateCfg)
 		if err != nil {
 			return fmt.Errorf("unpacking template config fails: %v", err)
 		}
@@ -741,8 +751,82 @@ func (b *Beat) registerTemplateLoading() error {
 			return err
 		}
 
-		if esCfg.Index != "" && (cfg.Name == "" || cfg.Pattern == "") && (b.Config.Template == nil || b.Config.Template.Enabled()) {
-			return fmt.Errorf("setup.template.name and setup.template.pattern have to be set if index name is modified.")
+		if esCfg.Index != "" &&
+			(templateCfg.Name == "" || templateCfg.Pattern == "") &&
+			(b.Config.Template == nil || b.Config.Template.Enabled()) {
+			return errors.New("setup.template.name and setup.template.pattern have to be set if index name is modified")
+		}
+
+		if b.Config.ILM.Enabled() {
+			cfgwarn.Beta("Index lifecycle management is enabled which is in beta.")
+
+			ilmCfg, err := getILMConfig(b)
+			if err != nil {
+				return err
+			}
+
+			// In case no template settings are set, config must be created
+			if b.Config.Template == nil {
+				b.Config.Template = common.NewConfig()
+			}
+			// Template name and pattern can't be configure when using ILM
+			logp.Info("Set setup.template.name to '%s' as ILM is enabled.", ilmCfg.RolloverAlias)
+			err = b.Config.Template.SetString("name", -1, ilmCfg.RolloverAlias)
+			if err != nil {
+				return errw.Wrap(err, "error setting setup.template.name")
+			}
+			pattern := fmt.Sprintf("%s-*", ilmCfg.RolloverAlias)
+			logp.Info("Set setup.template.pattern to '%s' as ILM is enabled.", pattern)
+			err = b.Config.Template.SetString("pattern", -1, pattern)
+			if err != nil {
+				return errw.Wrap(err, "error setting setup.template.pattern")
+			}
+
+			// rollover_alias and lifecycle.name can't be configured and will be overwritten
+			logp.Info("Set settings.index.lifecycle.rollover_alias in template to %s as ILM is enabled.", ilmCfg.RolloverAlias)
+			err = b.Config.Template.SetString("settings.index.lifecycle.rollover_alias", -1, ilmCfg.RolloverAlias)
+			if err != nil {
+				return errw.Wrap(err, "error setting settings.index.lifecycle.rollover_alias")
+			}
+			logp.Info("Set settings.index.lifecycle.name in template to %s as ILM is enabled.", ILMPolicyName)
+			err = b.Config.Template.SetString("settings.index.lifecycle.name", -1, ILMPolicyName)
+			if err != nil {
+				return errw.Wrap(err, "error setting settings.index.lifecycle.name")
+			}
+
+			// Set the ingestion index to the rollover alias
+			logp.Info("Set output.elasticsearch.index to '%s' as ILM is enabled.", ilmCfg.RolloverAlias)
+			esCfg.Index = ilmCfg.RolloverAlias
+			err = b.Config.Output.Config().SetString("index", -1, ilmCfg.RolloverAlias)
+			if err != nil {
+				return errw.Wrap(err, "error setting output.elasticsearch.index")
+			}
+
+			writeAliasCallback, err := b.writeAliasLoadingCallback()
+			if err != nil {
+				return err
+			}
+
+			// Load write alias already on
+			esConfig := b.Config.Output.Config()
+
+			// Check that ILM is enabled and the right elasticsearch version exists
+			esClient, err := elasticsearch.NewConnectedClient(esConfig)
+			if err != nil {
+				return err
+			}
+
+			err = checkElasticsearchVersionIlm(esClient)
+			if err != nil {
+				return err
+			}
+
+			err = checkILMFeatureEnabled(esClient)
+			if err != nil {
+				return err
+			}
+
+			elasticsearch.RegisterConnectCallback(writeAliasCallback)
 		}
 
 		if b.Config.Template == nil || (b.Config.Template != nil && b.Config.Template.Enabled()) {
@@ -754,6 +838,8 @@ func (b *Beat) registerTemplateLoading() error {
 				return err
 			}
 			elasticsearch.RegisterConnectCallback(callback)
+		} else if b.Config.ILM.Enabled() {
+			return errors.New("templates cannot be disable when using ILM")
 		}
 	}