otel: fix remaining tests to use in process testing collector

Author: mauri870

x-pack/filebeat/input/gcppubsub/otel_test.go

@@ -4,7 +4,7 @@
 
 //go:build integration && !agentbeat
 
-package gcppubsub
+package gcppubsub_test
 
 import (
 	"bytes"
@@ -20,6 +20,8 @@ import (
 
 	"github.com/elastic/beats/v7/libbeat/otelbeat/oteltest"
 	"github.com/elastic/beats/v7/libbeat/tests/integration"
+	"github.com/elastic/beats/v7/x-pack/filebeat/input/gcppubsub/testutil"
+	"github.com/elastic/beats/v7/x-pack/libbeat/common/otelbeat/oteltestcol"
 
 	"github.com/elastic/elastic-agent-libs/testing/estools"
 )
@@ -28,17 +30,17 @@ func TestGCPInputOTelE2E(t *testing.T) {
 	integration.EnsureESIsRunning(t)
 
 	// Create pubsub client for setting up and communicating to emulator.
-	client, clientCancel := testSetup(t)
+	client, clientCancel := testutil.TestSetup(t)
 	defer func() {
 		clientCancel()
 		client.Close()
 	}()
 
-	createTopic(t, client)
-	createSubscription(t, "test-subscription-otel", client)
-	createSubscription(t, "test-subscription-fb", client)
+	testutil.CreateTopic(t, client)
+	testutil.CreateSubscription(t, "test-subscription-otel", client)
+	testutil.CreateSubscription(t, "test-subscription-fb", client)
 	const numMsgs = 10
-	publishMessages(t, client, numMsgs)
+	testutil.PublishMessages(t, client, numMsgs)
 
 	host := integration.GetESURL(t, "http")
 	user := host.User.Username()
@@ -59,7 +61,7 @@ func TestGCPInputOTelE2E(t *testing.T) {
 		Subscription string
 	}
 
-	gcpConfig := `filebeat.inputs:
+	gcpFilebeatConfig := `filebeat.inputs:
 - type: gcp-pubsub
   project_id: test-project-id
   topic: test-topic-foo
@@ -83,13 +85,76 @@ processors:
     - add_kubernetes_metadata: ~
 `
 
-	// start filebeat in otel mode
-	filebeatOTel := integration.NewBeat(
-		t,
-		"filebeat-otel",
-		"../../filebeat.test",
-		"otel",
-	)
+	gcpOTelConfig := `exporters:
+    elasticsearch:
+        auth:
+            authenticator: beatsauth
+        compression: gzip
+        compression_params:
+            level: 1
+        endpoints:
+            - {{ .ESURL }}
+        logs_dynamic_pipeline:
+            enabled: true
+        logs_index: logs-integration-{{ .Namespace }}
+        mapping:
+            mode: bodymap
+        max_conns_per_host: 1
+        password: {{ .Password }}
+        retry:
+            enabled: true
+            initial_interval: 1s
+            max_interval: 1m0s
+            max_retries: 3
+        sending_queue:
+            batch:
+                flush_timeout: 10s
+                max_size: 1600
+                min_size: 0
+                sizer: items
+            block_on_overflow: true
+            enabled: true
+            num_consumers: 1
+            queue_size: 3200
+            wait_for_result: true
+        user: {{ .Username }}
+extensions:
+    beatsauth:
+        idle_connection_timeout: 3s
+        proxy_disable: false
+        timeout: 1m30s
+receivers:
+    filebeatreceiver:
+        filebeat:
+            inputs:
+                - credentials_file: "testdata/fake.json"
+                  project_id: test-project-id
+                  subscription:
+                    name: {{ .Subscription }}
+                  topic: test-topic-foo
+                  type: gcp-pubsub
+        output:
+            otelconsumer:
+        processors:
+            - add_host_metadata: ~
+            - add_cloud_metadata: ~
+            - add_docker_metadata: ~
+            - add_kubernetes_metadata: ~
+        queue.mem.flush.timeout: 0s
+        setup.template.enabled: false
+service:
+    extensions:
+        - beatsauth
+    pipelines:
+        logs:
+            exporters:
+                - elasticsearch
+            receivers:
+                - filebeatreceiver
+    telemetry:
+        metrics:
+            level: none
+`
 
 	optionsValue := options{
 		ESURL:    fmt.Sprintf("%s://%s", host.Scheme, host.Host),
@@ -100,19 +165,16 @@ processors:
 	var configBuffer bytes.Buffer
 	optionsValue.Namespace = otelNamespace
 	optionsValue.Subscription = "test-subscription-otel"
-	require.NoError(t, template.Must(template.New("config").Parse(gcpConfig)).Execute(&configBuffer, optionsValue))
-
-	filebeatOTel.WriteConfigFile(configBuffer.String())
+	require.NoError(t, template.Must(template.New("config").Parse(gcpOTelConfig)).Execute(&configBuffer, optionsValue))
 
-	filebeatOTel.Start()
-	defer filebeatOTel.Stop()
+	oteltestcol.New(t, configBuffer.String())
 
 	// reset buffer
 	configBuffer.Reset()
 
 	optionsValue.Namespace = fbNameSpace
 	optionsValue.Subscription = "test-subscription-fb"
-	require.NoError(t, template.Must(template.New("config").Parse(gcpConfig)).Execute(&configBuffer, optionsValue))
+	require.NoError(t, template.Must(template.New("config").Parse(gcpFilebeatConfig)).Execute(&configBuffer, optionsValue))
 
 	// start filebeat
 	filebeat := integration.NewBeat(
@@ -181,5 +243,4 @@ processors:
 	}
 
 	oteltest.AssertMapsEqual(t, filebeatDoc, otelDoc, ignoredFields, "expected documents to be equal")
-
 }