Move HTTP calls to Logstash from New() to Fetch() (elastic#15306) (el…

…astic#15323)

* node_stats: move HTTP calls from New() to Fetch()

* node: move HTTP calls from New() to Fetch()

* Adding CHANGELOG entry

* Refactoring: extract common code into logstash package

* Unexporting getVersion

CHANGELOG.next.asciidoc

@@ -68,6 +68,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fixed bug with `elasticsearch/cluster_stats` metricset not recording license ID in the correct field. {pull}14592[14592]
 - Fix `docker.container.size` fields values {issue}14979[14979] {pull}15224[15224]
 - Make `kibana` module more resilient to Kibana unavailability. {issue}15258[15258] {pull}15270[15270]
+- Make `logstash` module more resilient to Logstash unavailability. {issue}15276[15276] {pull}15306[15306]
 
 *Packetbeat*
 

metricbeat/module/logstash/logstash.go

@@ -19,6 +19,7 @@ package logstash
 
 import (
 	"encoding/json"
+	"fmt"
 	"net/url"
 
 	"github.com/pkg/errors"
@@ -158,8 +159,25 @@ func GetPipelines(m *MetricSet) ([]PipelineState, error) {
 	return pipelines, nil
 }
 
-// GetVersion returns the version of the Logstash node
-func GetVersion(m *MetricSet) (*common.Version, error) {
+// CheckPipelineGraphAPIsAvailable returns an error if pipeline graph APIs are not
+// available in the version of the Logstash node.
+func (m *MetricSet) CheckPipelineGraphAPIsAvailable() error {
+	logstashVersion, err := m.getVersion()
+	if err != nil {
+		return err
+	}
+
+	arePipelineGraphAPIsAvailable := elastic.IsFeatureAvailable(logstashVersion, PipelineGraphAPIsAvailableVersion)
+
+	if !arePipelineGraphAPIsAvailable {
+		const errorMsg = "the %v metricset with X-Pack enabled is only supported with Logstash >= %v. You are currently running Logstash %v"
+		return fmt.Errorf(errorMsg, m.FullyQualifiedName(), PipelineGraphAPIsAvailableVersion, logstashVersion)
+	}
+
+	return nil
+}
+
+func (m *MetricSet) getVersion() (*common.Version, error) {
 	const rootPath = "/"
 	content, err := fetchPath(m.HTTP, rootPath, "")
 	if err != nil {
@@ -178,12 +196,6 @@ func GetVersion(m *MetricSet) (*common.Version, error) {
 	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(httpHelper *helper.HTTP, path string, query string) ([]byte, error) {
 	currentURI := httpHelper.GetURI()
 	defer httpHelper.SetURI(currentURI)

metricbeat/module/logstash/node/node.go

@@ -18,8 +18,6 @@
 package node
 
 import (
-	"fmt"
-
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/mb/parse"
 	"github.com/elastic/beats/metricbeat/module/logstash"
@@ -58,23 +56,6 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 		return nil, err
 	}
 
-	if ms.XPack {
-		logstashVersion, err := logstash.GetVersion(ms)
-		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,
 	}, nil
@@ -106,3 +87,11 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) error {
 
 	return nil
 }
+
+func (m *MetricSet) init() error {
+	if m.XPack {
+		return m.CheckPipelineGraphAPIsAvailable()
+	}
+
+	return nil
+}

metricbeat/module/logstash/node_stats/node_stats.go

@@ -18,8 +18,6 @@
 package node_stats
 
 import (
-	"fmt"
-
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/mb/parse"
 	"github.com/elastic/beats/metricbeat/module/logstash"
@@ -59,25 +57,6 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 		return nil, err
 	}
 
-	if ms.XPack {
-		logstashVersion, err := logstash.GetVersion(ms)
-		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)
-		}
-
-		ms.HTTP.SetURI(ms.HTTP.GetURI() + "?vertices=true")
-	}
-
 	return &MetricSet{
 		ms,
 	}, nil
@@ -87,6 +66,15 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 // It returns the event which is then forward to the output. In case of an error, a
 // descriptive error must be returned.
 func (m *MetricSet) Fetch(r mb.ReporterV2) error {
+	err := m.init()
+	if err != nil {
+		if m.XPack {
+			m.Logger().Error(err)
+			return nil
+		}
+		return err
+	}
+
 	content, err := m.HTTP.FetchContent()
 	if err != nil {
 		if m.XPack {
@@ -107,3 +95,16 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) error {
 
 	return nil
 }
+
+func (m *MetricSet) init() error {
+	if m.XPack {
+		err := m.CheckPipelineGraphAPIsAvailable()
+		if err != nil {
+			return err
+		}
+
+		m.HTTP.SetURI(m.HTTP.GetURI() + "?vertices=true")
+	}
+
+	return nil
+}