diff --git a/x-pack/filebeat/tests/integration/otel_azure_blob_storage_test.go b/x-pack/filebeat/tests/integration/otel_azure_blob_storage_test.go new file mode 100644 index 00000000000..0b17edf5f79 --- /dev/null +++ b/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") +} diff --git a/x-pack/filebeat/tests/integration/otel_filebeat_input_test.go b/x-pack/filebeat/tests/integration/otel_filebeat_input_test.go index 0291631f8d0..7647bc47efc 100644 --- a/x-pack/filebeat/tests/integration/otel_filebeat_input_test.go +++ b/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 diff --git a/x-pack/filebeat/tests/integration/otel_gcppubsub_test.go b/x-pack/filebeat/tests/integration/otel_gcppubsub_test.go index 73565776d13..0fcfa3e7259 100644 --- a/x-pack/filebeat/tests/integration/otel_gcppubsub_test.go +++ b/x-pack/filebeat/tests/integration/otel_gcppubsub_test.go @@ -4,7 +4,7 @@ //go:build integration -package integration_test +package integration import ( "bytes" @@ -54,7 +54,7 @@ func TestGCPInputOTelE2E(t *testing.T) { fbIndex := "logs-integration-" + fbNameSpace type options struct { - Namespace string + Index string ESURL string Username string Password string @@ -75,7 +75,7 @@ output: - {{ .ESURL }} username: {{ .Username }} password: {{ .Password }} - index: logs-integration-{{ .Namespace }} + index: {{ .Index }} queue.mem.flush.timeout: 0s setup.template.enabled: false @@ -86,43 +86,7 @@ processors: - add_kubernetes_metadata: ~ ` - gcpOTelConfig := `exporters: - elasticsearch: - auth: - authenticator: beatsauth - compression: gzip - compression_params: - level: 1 - endpoints: - - {{ .ESURL }} - logs_dynamic_pipeline: - enabled: true - 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 -receivers: + gcpOTelConfig := otelElasticsearchExporterYAML + `receivers: filebeatreceiver: filebeat: inputs: @@ -140,19 +104,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 optionsValue := options{ ESURL: fmt.Sprintf("%s://%s", host.Scheme, host.Host), @@ -162,7 +114,7 @@ service: } var configBuffer bytes.Buffer - optionsValue.Namespace = otelNamespace + optionsValue.Index = otelIndex optionsValue.Subscription = "test-subscription-otel" require.NoError(t, template.Must(template.New("config").Parse(gcpOTelConfig)).Execute(&configBuffer, optionsValue)) @@ -171,7 +123,7 @@ service: // reset buffer configBuffer.Reset() - optionsValue.Namespace = fbNameSpace + optionsValue.Index = fbIndex optionsValue.Subscription = "test-subscription-fb" require.NoError(t, template.Must(template.New("config").Parse(gcpFilebeatConfig)).Execute(&configBuffer, optionsValue)) @@ -190,11 +142,7 @@ service: es := integration.GetESClient(t, "http") t.Cleanup(func() { - _, err := es.Indices.DeleteDataStream([]string{ - otelIndex, - fbIndex, - }) - require.NoError(t, err, "failed to delete indices") + deleteDataStreamsFromES(t, es, []string{otelIndex, fbIndex}) }) rawQuery := map[string]any{ diff --git a/x-pack/filebeat/tests/integration/otel_helpers_test.go b/x-pack/filebeat/tests/integration/otel_helpers_test.go new file mode 100644 index 00000000000..14f4fc09327 --- /dev/null +++ b/x-pack/filebeat/tests/integration/otel_helpers_test.go @@ -0,0 +1,98 @@ +// 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 ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/elastic/go-elasticsearch/v8" +) + +func deleteDataStreamsFromES(t *testing.T, es *elasticsearch.Client, dataStreams []string) { + t.Helper() + + _, err := es.Indices.DeleteDataStream(dataStreams) + require.NoError(t, err, "failed to delete data streams") +} + +const otelElasticsearchExporterYAML = `exporters: + elasticsearch: + auth: + authenticator: beatsauth + compression: gzip + compression_params: + level: 1 + endpoints: + - {{ .ESURL }} + logs_dynamic_pipeline: + enabled: true + 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 +` + +const otelElasticsearchServiceYAML = `service: + extensions: + - beatsauth + pipelines: + logs: + exporters: + - elasticsearch + receivers: + - filebeatreceiver + telemetry: + metrics: + level: none +` + +func otelE2ERawQueryForInputTypeAndMessage(inputType, message string) map[string]any { + return map[string]any{ + "query": map[string]any{ + "bool": map[string]any{ + "must": []map[string]any{ + { + "match_phrase": map[string]any{ + "input.type": inputType, + }, + }, + { + "match_phrase": map[string]any{ + "message": message, + }, + }, + }, + }, + }, + "sort": []map[string]any{ + {"@timestamp": map[string]any{"order": "asc"}}, + }, + } +} diff --git a/x-pack/filebeat/tests/integration/otel_kafka_test.go b/x-pack/filebeat/tests/integration/otel_kafka_test.go index 1955e9b301a..4ba646b38ed 100644 --- a/x-pack/filebeat/tests/integration/otel_kafka_test.go +++ b/x-pack/filebeat/tests/integration/otel_kafka_test.go @@ -27,7 +27,6 @@ import ( "github.com/elastic/beats/v7/x-pack/otel/oteltestcol" "github.com/elastic/elastic-agent-libs/mapstr" "github.com/elastic/elastic-agent-libs/testing/estools" - "github.com/elastic/go-elasticsearch/v8" "github.com/elastic/sarama" ) @@ -256,41 +255,7 @@ processors: - add_kubernetes_metadata: ~ ` - kafkaOTelConfig := `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 -receivers: + kafkaOTelConfig := otelElasticsearchExporterYAML + `receivers: filebeatreceiver: filebeat: inputs: @@ -311,19 +276,7 @@ receivers: setup.template.enabled: false management.otel.enabled: true path.home: {{ .PathHome }} -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), @@ -365,27 +318,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": "kafka", - }, - }, - { - "match_phrase": map[string]any{ - "message": kafkaInputTestMsg, - }, - }, - }, - }, - }, - "sort": []map[string]any{ - {"@timestamp": map[string]any{"order": "asc"}}, - }, - } + rawQuery := otelE2ERawQueryForInputTypeAndMessage("kafka", kafkaInputTestMsg) var filebeatDocs estools.Documents var otelDocs estools.Documents @@ -432,8 +365,3 @@ func deleteKafkaInputTopic(t *testing.T, topic string) { t.Logf("failed to delete topic %q: %v", topic, err) } } - -func deleteDataStreamsFromES(t *testing.T, es *elasticsearch.Client, dataStreams []string) { - _, err := es.Indices.DeleteDataStream(dataStreams) - require.NoError(t, err, "failed to delete data streams") -} diff --git a/x-pack/filebeat/tests/integration/otel_tcp_udp_test.go b/x-pack/filebeat/tests/integration/otel_tcp_udp_test.go index 6bf6898fefd..8168791e1a4 100644 --- a/x-pack/filebeat/tests/integration/otel_tcp_udp_test.go +++ b/x-pack/filebeat/tests/integration/otel_tcp_udp_test.go @@ -119,41 +119,7 @@ processors: - add_kubernetes_metadata: ~ ` - otelConfig := `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 -receivers: + otelConfig := otelElasticsearchExporterYAML + `receivers: filebeatreceiver: filebeat: inputs: @@ -168,19 +134,7 @@ receivers: queue.mem.flush.timeout: 0s setup.template.enabled: false path.home: {{ .PathHome }} -service: - extensions: - - beatsauth - pipelines: - logs: - exporters: - - elasticsearch - receivers: - - filebeatreceiver - telemetry: - metrics: - level: none -` +` + otelElasticsearchServiceYAML optionsValue := options{ InputType: inputType, @@ -226,34 +180,10 @@ service: es := integration.GetESClient(t, "http") t.Cleanup(func() { - _, err := es.Indices.DeleteDataStream([]string{ - otelIndex, - fbIndex, - }) - require.NoError(t, err, "failed to delete data streams") + deleteDataStreamsFromES(t, es, []string{otelIndex, fbIndex}) }) - rawQuery := map[string]any{ - "query": map[string]any{ - "bool": map[string]any{ - "must": []map[string]any{ - { - "match_phrase": map[string]any{ - "input.type": inputType, - }, - }, - { - "match_phrase": map[string]any{ - "message": testMessage, - }, - }, - }, - }, - }, - "sort": []map[string]any{ - {"@timestamp": map[string]any{"order": "asc"}}, - }, - } + rawQuery := otelE2ERawQueryForInputTypeAndMessage(inputType, testMessage) var filebeatDocs estools.Documents var otelDocs estools.Documents