Adding integration test for logstash Metricbeat module, xpack code pa…

…th (#15800)

* Adding integration test for logstash module, xpack data path

* Adding godoc

* Refactoring

* More refactoring

* Fixing reference

metricbeat/mb/testing/modules.go

@@ -86,6 +86,22 @@ func NewTestModule(t testing.TB, config interface{}) *TestModule {
 // The ModuleFactory and MetricSetFactory are obtained from the global
 // Registry.
 func NewMetricSet(t testing.TB, config interface{}) mb.MetricSet {
+	metricsets := NewMetricSets(t, config)
+
+	if len(metricsets) != 1 {
+		t.Fatal("invalid number of metricsets instantiated")
+	}
+
+	metricset := metricsets[0]
+	if metricset == nil {
+		t.Fatal("metricset is nil")
+	}
+	return metricset
+}
+
+// NewMetricSets instantiates a list of new MetricSets using the given
+// module configuration.
+func NewMetricSets(t testing.TB, config interface{}) []mb.MetricSet {
 	c, err := common.NewConfigFrom(config)
 	if err != nil {
 		t.Fatal(err)
@@ -98,15 +114,7 @@ func NewMetricSet(t testing.TB, config interface{}) mb.MetricSet {
 		t.Fatal("no module instantiated")
 	}
 
-	if len(metricsets) != 1 {
-		t.Fatal("invalid number of metricsets instantiated")
-	}
-
-	metricset := metricsets[0]
-	if metricset == nil {
-		t.Fatal("metricset is nil")
-	}
-	return metricset
+	return metricsets
 }
 
 // NewEventFetcher instantiates a new EventFetcher using the given
@@ -182,6 +190,22 @@ func NewReportingMetricSetV2Error(t testing.TB, config interface{}) mb.Reporting
 	return reportingMetricSetV2Error
 }
 
+// NewReportingMetricSetV2Errors returns an array of new ReportingMetricSetV2 instances.
+func NewReportingMetricSetV2Errors(t testing.TB, config interface{}) []mb.ReportingMetricSetV2Error {
+	metricSets := NewMetricSets(t, config)
+	var reportingMetricSets []mb.ReportingMetricSetV2Error
+	for _, metricSet := range metricSets {
+		rMS, ok := metricSet.(mb.ReportingMetricSetV2Error)
+		if !ok {
+			t.Fatalf("MetricSet %v does not implement ReportingMetricSetV2Error", metricSet.Name())
+		}
+
+		reportingMetricSets = append(reportingMetricSets, rMS)
+	}
+
+	return reportingMetricSets
+}
+
 // NewReportingMetricSetV2WithContext returns a new ReportingMetricSetV2WithContext instance. Then
 // you can use ReportingFetchV2 to perform a Fetch operation with the MetricSet.
 func NewReportingMetricSetV2WithContext(t testing.TB, config interface{}) mb.ReportingMetricSetV2WithContext {

metricbeat/module/logstash/logstash_integration_test.go

@@ -41,7 +41,7 @@ func TestFetch(t *testing.T) {
 
 	for _, metricSet := range metricSets {
 		t.Run(metricSet, func(t *testing.T) {
-			config := logstash.GetConfig(metricSet, service.Host())
+			config := getConfig(metricSet, service.Host())
 			f := mbtest.NewReportingMetricSetV2Error(t, config)
 			events, errs := mbtest.ReportingFetchV2Error(f)
 
@@ -57,11 +57,11 @@ func TestFetch(t *testing.T) {
 }
 
 func TestData(t *testing.T) {
-	service := compose.EnsureUp(t, "logstash")
+	service := compose.EnsureUpWithTimeout(t, 300, "logstash")
 
 	for _, metricSet := range metricSets {
 		t.Run(metricSet, func(t *testing.T) {
-			config := logstash.GetConfig(metricSet, service.Host())
+			config := getConfig(metricSet, service.Host())
 			f := mbtest.NewReportingMetricSetV2Error(t, config)
 			err := mbtest.WriteEventsReporterV2Error(f, t, metricSet)
 			if err != nil {
@@ -70,3 +70,44 @@ func TestData(t *testing.T) {
 		})
 	}
 }
+
+func TestXPackEnabled(t *testing.T) {
+	service := compose.EnsureUpWithTimeout(t, 300, "logstash")
+
+	metricSetToTypeMap := map[string]string{
+		"node":       "logstash_state",
+		"node_stats": "logstash_stats",
+	}
+
+	config := getXPackConfig(service.Host())
+
+	metricSets := mbtest.NewReportingMetricSetV2Errors(t, config)
+	for _, metricSet := range metricSets {
+		events, errs := mbtest.ReportingFetchV2Error(metricSet)
+		assert.Empty(t, errs)
+		if !assert.NotEmpty(t, events) {
+			t.FailNow()
+		}
+
+		event := events[0]
+		assert.Equal(t, metricSetToTypeMap[metricSet.Name()], event.RootFields["type"])
+		assert.Regexp(t, `^.monitoring-logstash-\d-mb`, event.Index)
+	}
+}
+
+func getConfig(metricSet string, host string) map[string]interface{} {
+	return map[string]interface{}{
+		"module":     logstash.ModuleName,
+		"metricsets": []string{metricSet},
+		"hosts":      []string{host},
+	}
+}
+
+func getXPackConfig(host string) map[string]interface{} {
+	return map[string]interface{}{
+		"module":        logstash.ModuleName,
+		"metricsets":    metricSets,
+		"hosts":         []string{host},
+		"xpack.enabled": true,
+	}
+}