Adding version check to Logstash metricsets when xpack.enabled: true is set

metricbeat/module/logstash/logstash.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/metricbeat/helper"
+	"github.com/elastic/beats/metricbeat/helper/elastic"
 	"github.com/elastic/beats/metricbeat/mb"
 )
 
@@ -74,6 +75,10 @@ func validateXPackMetricsets(base mb.BaseModule) error {
 // ModuleName is the name of this module.
 const ModuleName = "logstash"
 
+// PipelineGraphAPIsAvailableVersion is the version of Logstash since when its APIs
+// can return pipeline graphs
+var PipelineGraphAPIsAvailableVersion = common.MustNewVersion("7.3.0")
+
 // MetricSet can be used to build other metricsets within the Logstash module.
 type MetricSet struct {
 	mb.BaseMetricSet
@@ -135,6 +140,32 @@ func GetPipelines(http *helper.HTTP, resetURI string) ([]PipelineState, error) {
 	return pipelines, nil
 }
 
+// GetVersion returns the version of the Logstash node
+func GetVersion(http *helper.HTTP, currentPath string) (*common.Version, error) {
+	const rootPath = "/"
+	content, err := fetchPath(http, currentPath, rootPath, "")
+	if err != nil {
+		return nil, err
+	}
+
+	var response struct {
+		Version *common.Version `json:"version"`
+	}
+
+	err = json.Unmarshal(content, &response)
+	if err != nil {
+		return nil, err
+	}
+
+	return response.Version, nil
+}
+
+// ArePipelineGraphAPIsAvailable returns whether Logstash APIs that returns pipeline graphs
+// are available in the given version of Logstash
+func ArePipelineGraphAPIsAvailable(currentLogstashVersion *common.Version) bool {
+	return elastic.IsFeatureAvailable(currentLogstashVersion, PipelineGraphAPIsAvailableVersion)
+}
+
 func fetchPath(http *helper.HTTP, resetURI, path string, query string) ([]byte, error) {
 	defer http.SetURI(resetURI)
 

metricbeat/module/logstash/node/node.go

@@ -18,6 +18,8 @@
 package node
 
 import (
+	"fmt"
+
 	"github.com/elastic/beats/metricbeat/helper"
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/mb/parse"
@@ -63,6 +65,23 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 		return nil, err
 	}
 
+	if ms.XPack {
+		logstashVersion, err := logstash.GetVersion(http, ms.HostData().SanitizedURI+nodePath)
+		if err != nil {
+			return nil, err
+		}
+
+		arePipelineGraphAPIsAvailable := logstash.ArePipelineGraphAPIsAvailable(logstashVersion)
+		if err != nil {
+			return nil, err
+		}
+
+		if !arePipelineGraphAPIsAvailable {
+			const errorMsg = "The %v metricset with X-Pack enabled is only supported with Logstash >= %v. You are currently running Logstash %v"
+			return nil, fmt.Errorf(errorMsg, ms.FullyQualifiedName(), logstash.PipelineGraphAPIsAvailableVersion, logstashVersion)
+		}
+	}
+
 	return &MetricSet{
 		ms,
 		http,

metricbeat/module/logstash/node_stats/node_stats.go

@@ -18,6 +18,8 @@
 package node_stats
 
 import (
+	"fmt"
+
 	"github.com/elastic/beats/metricbeat/helper"
 
 	"github.com/elastic/beats/metricbeat/mb"
@@ -66,6 +68,21 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	}
 
 	if ms.XPack {
+		logstashVersion, err := logstash.GetVersion(http, ms.HostData().SanitizedURI+nodeStatsPath)
+		if err != nil {
+			return nil, err
+		}
+
+		arePipelineGraphAPIsAvailable := logstash.ArePipelineGraphAPIsAvailable(logstashVersion)
+		if err != nil {
+			return nil, err
+		}
+
+		if !arePipelineGraphAPIsAvailable {
+			const errorMsg = "The %v metricset with X-Pack enabled is only supported with Logstash >= %v. You are currently running Logstash %v"
+			return nil, fmt.Errorf(errorMsg, ms.FullyQualifiedName(), logstash.PipelineGraphAPIsAvailableVersion, logstashVersion)
+		}
+
 		http.SetURI(http.GetURI() + "?vertices=true")
 	}