[beatreceiver] Add e2e test for azure blob storage (#51738)

* [beatreceiver] Add e2e test for azure blob storage

Author: khushijain21

x-pack/filebeat/tests/integration/otel_azure_blob_storage_test.go

@@ -0,0 +1,205 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+//go:build integration
+
+package integration
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"text/template"
+	"time"
+
+	"github.com/gofrs/uuid/v5"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/beats/v7/libbeat/tests/integration"
+	"github.com/elastic/beats/v7/x-pack/filebeat/input/azureblobstorage/mock"
+	"github.com/elastic/beats/v7/x-pack/otel/oteltest"
+	"github.com/elastic/beats/v7/x-pack/otel/oteltestcol"
+	"github.com/elastic/elastic-agent-libs/testing/estools"
+)
+
+const (
+	azureBlobTestAccountName = "beatsblobnew"
+	azureBlobTestAccountKey  = "7pfLm1betGiRyyABEM/RFrLYlafLZHbLtGhB52LkWVeBxE7la9mIvk6YYAbQKYE/f0GdhiaOZeV8+AStsAdr/Q=="
+	azureBlobTestContainer   = "beatscontainer"
+	azureBlobTestBlob        = "ata.json"
+	azureBlobTestMessage     = "iPhone 9"
+)
+
+func startAzureBlobMockStorageServer(t *testing.T) string {
+	t.Helper()
+
+	srv := httptest.NewServer(mock.AzureStorageServer())
+	t.Cleanup(srv.Close)
+
+	return srv.URL + "/"
+}
+
+func TestAzureBlobStorageInputOTelE2E(t *testing.T) {
+	integration.EnsureESIsRunning(t)
+
+	storageURL := startAzureBlobMockStorageServer(t)
+	otelHome := t.TempDir()
+
+	host := integration.GetESURL(t, "http")
+	user := host.User.Username()
+	password, _ := host.User.Password()
+
+	otelNamespace := strings.ReplaceAll(uuid.Must(uuid.NewV4()).String(), "-", "")
+	fbNamespace := strings.ReplaceAll(uuid.Must(uuid.NewV4()).String(), "-", "")
+
+	otelIndex := "logs-integration-" + otelNamespace
+	fbIndex := "logs-integration-" + fbNamespace
+
+	type options struct {
+		Index       string
+		ESURL       string
+		Username    string
+		Password    string
+		StorageURL  string
+		PathHome    string
+		AccountName string
+		AccountKey  string
+	}
+
+	filebeatConfig := `filebeat.inputs:
+- type: azure-blob-storage
+  id: azure-blob-storage-input-e2e
+  account_name: {{ .AccountName }}
+  storage_url: {{ .StorageURL }}
+  auth:
+    shared_credentials:
+      account_key: {{ .AccountKey }}
+  poll: false
+  max_workers: 1
+  containers:
+    - name: ` + azureBlobTestContainer + `
+  file_selectors:
+    - regex: '^` + azureBlobTestBlob + `$'
+
+output:
+  elasticsearch:
+    hosts:
+      - {{ .ESURL }}
+    username: {{ .Username }}
+    password: {{ .Password }}
+    index: {{ .Index }}
+
+queue.mem.flush.timeout: 0s
+setup.template.enabled: false
+processors:
+    - add_host_metadata: ~
+    - add_cloud_metadata: ~
+    - add_docker_metadata: ~
+    - add_kubernetes_metadata: ~
+`
+
+	otelConfig := otelElasticsearchExporterYAML + `receivers:
+    filebeatreceiver:
+        filebeat:
+            inputs:
+                - type: azure-blob-storage
+                  id: azure-blob-storage-input-e2e
+                  account_name: {{ .AccountName }}
+                  storage_url: {{ .StorageURL }}
+                  auth:
+                    shared_credentials:
+                      account_key: {{ .AccountKey }}
+                  poll: false
+                  max_workers: 1
+                  containers:
+                    - name: ` + azureBlobTestContainer + `
+                  file_selectors:
+                    - regex: '^` + azureBlobTestBlob + `$'
+        processors:
+            - add_host_metadata: ~
+            - add_cloud_metadata: ~
+            - add_docker_metadata: ~
+            - add_kubernetes_metadata: ~
+        queue.mem.flush.timeout: 0s
+        setup.template.enabled: false
+        path.home: {{ .PathHome }}
+` + otelElasticsearchServiceYAML
+
+	optionsValue := options{
+		ESURL:       fmt.Sprintf("%s://%s", host.Scheme, host.Host),
+		Username:    user,
+		Password:    password,
+		StorageURL:  storageURL,
+		PathHome:    otelHome,
+		AccountName: azureBlobTestAccountName,
+		AccountKey:  azureBlobTestAccountKey,
+	}
+
+	var configBuffer bytes.Buffer
+	optionsValue.Index = otelIndex
+	require.NoError(t, template.Must(template.New("config").Parse(otelConfig)).Execute(&configBuffer, optionsValue))
+
+	oteltestcol.New(t, configBuffer.String())
+
+	configBuffer.Reset()
+
+	optionsValue.Index = fbIndex
+	require.NoError(t, template.Must(template.New("config").Parse(filebeatConfig)).Execute(&configBuffer, optionsValue))
+
+	filebeat := integration.NewBeat(
+		t,
+		"filebeat",
+		"../../filebeat.test",
+	)
+	filebeat.WriteConfigFile(configBuffer.String())
+	filebeat.Start()
+	defer filebeat.Stop()
+
+	filebeat.WaitLogsContains(
+		"filebeat start running",
+		20*time.Second,
+		"filebeat did not run",
+	)
+
+	es := integration.GetESClient(t, "http")
+
+	t.Cleanup(func() {
+		deleteDataStreamsFromES(t, es, []string{otelIndex, fbIndex})
+	})
+
+	rawQuery := otelE2ERawQueryForInputTypeAndMessage("azure-blob-storage", azureBlobTestMessage)
+
+	var filebeatDocs estools.Documents
+	var otelDocs estools.Documents
+	var err error
+
+	require.EventuallyWithTf(t,
+		func(ct *assert.CollectT) {
+			findCtx, findCancel := context.WithTimeout(t.Context(), 900*time.Millisecond)
+			defer findCancel()
+
+			otelDocs, err = estools.PerformQueryForRawQuery(findCtx, rawQuery, ".ds-"+otelIndex+"*", es)
+			assert.NoError(ct, err)
+			assert.GreaterOrEqual(ct, otelDocs.Hits.Total.Value, 1, "expected at least 1 otel document, got %d", otelDocs.Hits.Total.Value)
+
+			filebeatDocs, err = estools.PerformQueryForRawQuery(findCtx, rawQuery, ".ds-"+fbIndex+"*", es)
+			assert.NoError(ct, err)
+			assert.GreaterOrEqual(ct, filebeatDocs.Hits.Total.Value, 1, "expected at least 1 filebeat document, got %d", filebeatDocs.Hits.Total.Value)
+		},
+		3*time.Minute, 1*time.Second, "expected at least 1 document for both filebeat and otel modes")
+
+	filebeatDoc := filebeatDocs.Hits.Hits[0].Source
+	otelDoc := otelDocs.Hits.Hits[0].Source
+	ignoredFields := []string{
+		"@timestamp",
+		"agent.ephemeral_id",
+		"agent.id",
+	}
+
+	oteltest.AssertMapsEqual(t, filebeatDoc, otelDoc, ignoredFields, "expected documents to be equal")
+}

x-pack/filebeat/tests/integration/otel_filebeat_input_test.go

@@ -51,7 +51,7 @@ func TestCELInputOTelE2E(t *testing.T) {
 	fbIndex := "logs-integration-" + fbNamespace
 
 	type options struct {
-		Namespace   string
+		Index       string
 		ESURL       string
 		Username    string
 		Password    string
@@ -72,7 +72,7 @@ output:
       - {{ .ESURL }}
     username: {{ .Username }}
     password: {{ .Password }}
-    index: logs-integration-{{ .Namespace }}
+    index: {{ .Index }}
 
 queue.mem.flush.timeout: 0s
 setup.template.enabled: false
@@ -83,40 +83,8 @@ processors:
     - add_kubernetes_metadata: ~
 `
 
-	celOTelConfig := `exporters:
-    elasticsearch:
-        auth:
-            authenticator: beatsauth
-        compression: gzip
-        compression_params:
-            level: 1
-        endpoints:
-            - {{ .ESURL }}
-        logs_index: logs-integration-{{ .Namespace }}
-        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
+	celOTelConfig := otelElasticsearchExporterYAML +
+		`
 receivers:
     filebeatreceiver:
         filebeat:
@@ -134,19 +102,7 @@ receivers:
         queue.mem.flush.timeout: 0s
         setup.template.enabled: false
         management.otel.enabled: true
-service:
-    extensions:
-        - beatsauth
-    pipelines:
-        logs:
-            exporters:
-                - elasticsearch
-            receivers:
-                - filebeatreceiver
-    telemetry:
-        metrics:
-            level: none
-`
+` + otelElasticsearchServiceYAML
 
 	var configBuffer bytes.Buffer
 	require.NoError(t, template.Must(template.New("config").Parse(celOTelConfig)).Execute(&configBuffer, options{
@@ -155,7 +111,7 @@ service:
 		Password:    password,
 		ResourceURL: celSrv.URL,
 		Program:     celProgram,
-		Namespace:   otelNamespace,
+		Index:       otelIndex,
 	}))
 
 	oteltestcol.New(t, configBuffer.String())
@@ -168,7 +124,7 @@ service:
 		Password:    password,
 		ResourceURL: celSrv.URL,
 		Program:     celProgram,
-		Namespace:   fbNamespace,
+		Index:       fbIndex,
 	}))
 
 	filebeat := integration.NewBeat(
@@ -690,40 +646,7 @@ processors:
     - add_kubernetes_metadata: ~
 `
 
-	mqttOTelConfig := `exporters:
-    elasticsearch:
-        auth:
-            authenticator: beatsauth
-        compression: gzip
-        compression_params:
-            level: 1
-        endpoints:
-            - {{ .ESURL }}
-        logs_index: {{ .Index }}
-        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
+	mqttOTelConfig := otelElasticsearchExporterYAML + `
 receivers:
     filebeatreceiver:
         filebeat:
@@ -744,19 +667,7 @@ receivers:
         setup.template.enabled: false
         path.home: {{ .PathHome }}
         management.otel.enabled: true
-service:
-    extensions:
-        - beatsauth
-    pipelines:
-        logs:
-            exporters:
-                - elasticsearch
-            receivers:
-                - filebeatreceiver
-    telemetry:
-        metrics:
-            level: none
-`
+` + otelElasticsearchServiceYAML
 
 	optionsValue := options{
 		ESURL:    fmt.Sprintf("%s://%s", host.Scheme, host.Host),
@@ -798,27 +709,7 @@ service:
 		})
 	})
 
-	rawQuery := map[string]any{
-		"query": map[string]any{
-			"bool": map[string]any{
-				"must": []map[string]any{
-					{
-						"match_phrase": map[string]any{
-							"input.type": "mqtt",
-						},
-					},
-					{
-						"match_phrase": map[string]any{
-							"message": mqttInputTestMsg,
-						},
-					},
-				},
-			},
-		},
-		"sort": []map[string]any{
-			{"@timestamp": map[string]any{"order": "asc"}},
-		},
-	}
+	rawQuery := otelE2ERawQueryForInputTypeAndMessage("mqtt", mqttInputTestMsg)
 
 	var filebeatDocs estools.Documents
 	var otelDocs estools.Documents