Merge remote-tracking branch 'upstream/master' into feature/macosx-pipeline-2.0

* upstream/master:
  libbeat/cmd/instance: report cgroup stats (elastic#21113)
  Configurable index template loading (elastic#21212)
  [Ingest Manager] Thread safe sorted set (elastic#21290)

Author: v1v

CHANGELOG.next.asciidoc

@@ -437,6 +437,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Added experimental dataset `juniper/netscreen`. {pull}20820[20820]
 - Added experimental dataset `sophos/utm`. {pull}20820[20820]
 - Add Cloud Foundry tags in related events. {pull}21177[21177]
+- Add option to select the type of index template to load: legacy, component, index. {pull}21212[21212]
 
 *Auditbeat*
 

auditbeat/auditbeat.reference.yml

@@ -1144,6 +1144,11 @@ output.elasticsearch:
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default auditbeat uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "auditbeat-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "auditbeat-%{[agent.version]}"

filebeat/filebeat.reference.yml

@@ -1870,6 +1870,11 @@ output.elasticsearch:
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default filebeat uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "filebeat-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "filebeat-%{[agent.version]}"

heartbeat/heartbeat.reference.yml

@@ -1321,6 +1321,11 @@ output.elasticsearch:
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default heartbeat uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "heartbeat-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "heartbeat-%{[agent.version]}"

journalbeat/journalbeat.reference.yml

@@ -1086,6 +1086,11 @@ output.elasticsearch:
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default journalbeat uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "journalbeat-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "journalbeat-%{[agent.version]}"

libbeat/_meta/config/setup.template.reference.yml.tmpl

@@ -7,6 +7,11 @@
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default {{.BeatName}} uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "{{.BeatIndexPrefix}}-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "{{.BeatIndexPrefix}}-%{[agent.version]}"

libbeat/cmd/instance/metrics.go

@@ -28,6 +28,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/metric/system/cpu"
 	"github.com/elastic/beats/v7/libbeat/metric/system/process"
 	"github.com/elastic/beats/v7/libbeat/monitoring"
+	"github.com/elastic/gosigar/cgroup"
 )
 
 var (
@@ -65,10 +66,15 @@ func setupMetrics(name string) error {
 }
 
 func setupPlatformSpecificMetrics() {
+	switch runtime.GOOS {
+	case "linux":
+		monitoring.NewFunc(beatMetrics, "cgroup", reportBeatCgroups, monitoring.Report)
+	case "windows":
+		setupWindowsHandlesMetrics()
+	}
+
 	if runtime.GOOS != "windows" {
 		monitoring.NewFunc(systemMetrics, "load", reportSystemLoadAverage, monitoring.Report)
-	} else {
-		setupWindowsHandlesMetrics()
 	}
 
 	setupLinuxBSDFDMetrics()
@@ -254,3 +260,79 @@ func reportRuntime(_ monitoring.Mode, V monitoring.Visitor) {
 
 	monitoring.ReportInt(V, "goroutines", int64(runtime.NumGoroutine()))
 }
+
+func reportBeatCgroups(_ monitoring.Mode, V monitoring.Visitor) {
+	V.OnRegistryStart()
+	defer V.OnRegistryFinished()
+
+	pid, err := process.GetSelfPid()
+	if err != nil {
+		logp.Err("error getting PID for self process: %v", err)
+		return
+	}
+
+	cgroups, err := cgroup.NewReader("", true)
+	if err != nil {
+		if err == cgroup.ErrCgroupsMissing {
+			logp.Warn("cgroup data collection disabled: %v", err)
+		} else {
+			logp.Err("cgroup data collection disabled: %v", err)
+		}
+		return
+	}
+	selfStats, err := cgroups.GetStatsForProcess(pid)
+	if err != nil {
+		logp.Err("error getting group status: %v", err)
+		return
+	}
+
+	if cpu := selfStats.CPU; cpu != nil {
+		monitoring.ReportNamespace(V, "cpu", func() {
+			if cpu.ID != "" {
+				monitoring.ReportString(V, "id", cpu.ID)
+			}
+			monitoring.ReportNamespace(V, "cfs", func() {
+				monitoring.ReportNamespace(V, "period", func() {
+					monitoring.ReportInt(V, "us", int64(cpu.CFS.PeriodMicros))
+				})
+				monitoring.ReportNamespace(V, "quota", func() {
+					monitoring.ReportInt(V, "us", int64(cpu.CFS.QuotaMicros))
+				})
+			})
+			monitoring.ReportNamespace(V, "stats", func() {
+				monitoring.ReportInt(V, "periods", int64(cpu.Stats.Periods))
+				monitoring.ReportNamespace(V, "throttled", func() {
+					monitoring.ReportInt(V, "periods", int64(cpu.Stats.ThrottledPeriods))
+					monitoring.ReportInt(V, "ns", int64(cpu.Stats.ThrottledTimeNanos))
+				})
+			})
+		})
+	}
+
+	if cpuacct := selfStats.CPUAccounting; cpuacct != nil {
+		monitoring.ReportNamespace(V, "cpuacct", func() {
+			if cpuacct.ID != "" {
+				monitoring.ReportString(V, "id", cpuacct.ID)
+			}
+			monitoring.ReportNamespace(V, "total", func() {
+				monitoring.ReportInt(V, "ns", int64(cpuacct.TotalNanos))
+			})
+		})
+	}
+
+	if memory := selfStats.Memory; memory != nil {
+		monitoring.ReportNamespace(V, "memory", func() {
+			if memory.ID != "" {
+				monitoring.ReportString(V, "id", memory.ID)
+			}
+			monitoring.ReportNamespace(V, "mem", func() {
+				monitoring.ReportNamespace(V, "limit", func() {
+					monitoring.ReportInt(V, "bytes", int64(memory.Mem.Limit))
+				})
+				monitoring.ReportNamespace(V, "usage", func() {
+					monitoring.ReportInt(V, "bytes", int64(memory.Mem.Usage))
+				})
+			})
+		})
+	}
+}

libbeat/docs/template-config.asciidoc

@@ -26,6 +26,11 @@ existing one.
 *`setup.template.enabled`*:: Set to false to disable template loading. If set this to false,
 you must <<load-template-manually,load the template manually>>.
 
+*`setup.template.type`*:: The type of template to use. Available options: `legacy` (default), index templates
+before Elasticsearch v7.8. Use this to avoid breaking existing deployments. New options are `composite`
+and `index`. Selecting `component` loads a component template which can be included in new index templates.
+The option `index` loads the new index template.
+
 *`setup.template.name`*:: The name of the template. The default is
 +{beatname_lc}+. The {beatname_uc} version is always appended to the given
 name, so the final name is +{beatname_lc}-%{[{beat_version_key}]}+.

libbeat/template/config.go

@@ -17,7 +17,27 @@
 
 package template
 
-import "github.com/elastic/beats/v7/libbeat/mapping"
+import (
+	"fmt"
+
+	"github.com/elastic/beats/v7/libbeat/mapping"
+)
+
+const (
+	IndexTemplateLegacy IndexTemplateType = iota
+	IndexTemplateComponent
+	IndexTemplateIndex
+)
+
+var (
+	templateTypes = map[string]IndexTemplateType{
+		"legacy":    IndexTemplateLegacy,
+		"component": IndexTemplateComponent,
+		"index":     IndexTemplateIndex,
+	}
+)
+
+type IndexTemplateType uint8
 
 // TemplateConfig holds config information about the Elasticsearch template
 type TemplateConfig struct {
@@ -30,10 +50,12 @@ type TemplateConfig struct {
 		Path    string `config:"path"`
 		Name    string `config:"name"`
 	} `config:"json"`
-	AppendFields mapping.Fields   `config:"append_fields"`
-	Overwrite    bool             `config:"overwrite"`
-	Settings     TemplateSettings `config:"settings"`
-	Order        int              `config:"order"`
+	AppendFields mapping.Fields    `config:"append_fields"`
+	Overwrite    bool              `config:"overwrite"`
+	Settings     TemplateSettings  `config:"settings"`
+	Order        int               `config:"order"`
+	Priority     int               `config:"priority"`
+	Type         IndexTemplateType `config:"type"`
 }
 
 // TemplateSettings are part of the Elasticsearch template and hold index and source specific information.
@@ -45,8 +67,26 @@ type TemplateSettings struct {
 // DefaultConfig for index template
 func DefaultConfig() TemplateConfig {
 	return TemplateConfig{
-		Enabled: true,
-		Fields:  "",
-		Order:   1,
+		Enabled:  true,
+		Fields:   "",
+		Type:     IndexTemplateLegacy,
+		Order:    1,
+		Priority: 150,
 	}
 }
+
+func (t *IndexTemplateType) Unpack(v string) error {
+	if v == "" {
+		*t = IndexTemplateLegacy
+		return nil
+	}
+
+	var tt IndexTemplateType
+	var ok bool
+	if tt, ok = templateTypes[v]; !ok {
+		return fmt.Errorf("unknown index template type: %s", v)
+	}
+	*t = tt
+
+	return nil
+}

libbeat/template/load.go

@@ -31,6 +31,14 @@ import (
 	"github.com/elastic/beats/v7/libbeat/paths"
 )
 
+var (
+	templateLoaderPath = map[IndexTemplateType]string{
+		IndexTemplateLegacy:    "/_template/",
+		IndexTemplateComponent: "/_component_template/",
+		IndexTemplateIndex:     "/_index_template/",
+	}
+)
+
 //Loader interface for loading templates
 type Loader interface {
 	Load(config TemplateConfig, info beat.Info, fields []byte, migration bool) error
@@ -97,7 +105,7 @@ func (l *ESLoader) Load(config TemplateConfig, info beat.Info, fields []byte, mi
 		templateName = config.JSON.Name
 	}
 
-	if l.templateExists(templateName) && !config.Overwrite {
+	if l.templateExists(templateName, config.Type) && !config.Overwrite {
 		l.log.Infof("Template %s already exists and will not be overwritten.", templateName)
 		return nil
 	}
@@ -107,7 +115,7 @@ func (l *ESLoader) Load(config TemplateConfig, info beat.Info, fields []byte, mi
 	if err != nil {
 		return err
 	}
-	if err := l.loadTemplate(templateName, body); err != nil {
+	if err := l.loadTemplate(templateName, config.Type, body); err != nil {
 		return fmt.Errorf("could not load template. Elasticsearch returned: %v. Template is: %s", err, body.StringToPrint())
 	}
 	l.log.Infof("template with name '%s' loaded.", templateName)
@@ -117,10 +125,11 @@ func (l *ESLoader) Load(config TemplateConfig, info beat.Info, fields []byte, mi
 // loadTemplate loads a template into Elasticsearch overwriting the existing
 // template if it exists. If you wish to not overwrite an existing template
 // then use CheckTemplate prior to calling this method.
-func (l *ESLoader) loadTemplate(templateName string, template map[string]interface{}) error {
+func (l *ESLoader) loadTemplate(templateName string, templateType IndexTemplateType, template map[string]interface{}) error {
 	l.log.Infof("Try loading template %s to Elasticsearch", templateName)
-	path := "/_template/" + templateName
-	params := esVersionParams(l.client.GetVersion())
+	clientVersion := l.client.GetVersion()
+	path := templateLoaderPath[templateType] + templateName
+	params := esVersionParams(clientVersion)
 	status, body, err := l.client.Request("PUT", path, "", params, template)
 	if err != nil {
 		return fmt.Errorf("couldn't load template: %v. Response body: %s", err, body)
@@ -133,11 +142,16 @@ func (l *ESLoader) loadTemplate(templateName string, template map[string]interfa
 
 // templateExists checks if a given template already exist. It returns true if
 // and only if Elasticsearch returns with HTTP status code 200.
-func (l *ESLoader) templateExists(templateName string) bool {
+func (l *ESLoader) templateExists(templateName string, templateType IndexTemplateType) bool {
 	if l.client == nil {
 		return false
 	}
 
+	if templateType == IndexTemplateComponent {
+		status, _, _ := l.client.Request("GET", "/_component_template/"+templateName, "", nil, nil)
+		return status == http.StatusOK
+	}
+
 	status, body, _ := l.client.Request("GET", "/_cat/templates/"+templateName, "", nil, nil)
 
 	return status == http.StatusOK && strings.Contains(string(body), templateName)