diff --git a/metricbeat/mb/testing/modules.go b/metricbeat/mb/testing/modules.go
index 12ff0e9a1ad..3edd1baeef9 100644
--- a/metricbeat/mb/testing/modules.go
+++ b/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 {
diff --git a/metricbeat/module/logstash/logstash_integration_test.go b/metricbeat/module/logstash/logstash_integration_test.go
index 15eb637faf6..fe9c54e46d7 100644
--- a/metricbeat/module/logstash/logstash_integration_test.go
+++ b/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,
+	}
+}
diff --git a/metricbeat/module/logstash/testing.go b/metricbeat/module/logstash/testing.go
deleted file mode 100644
index b63ce5cc5c2..00000000000
--- a/metricbeat/module/logstash/testing.go
+++ /dev/null
@@ -1,27 +0,0 @@
-// 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 logstash
-
-// GetConfig for Logstash
-func GetConfig(metricset string, host string) map[string]interface{} {
-	return map[string]interface{}{
-		"module":     ModuleName,
-		"metricsets": []string{metricset},
-		"hosts":      []string{host},
-	}
-}