[beatreceiver] Add e2e test to compare docs from redis and mqtt input

[khushijain21]

Proposed commit message

This PR adds e2e test that compares document ingested by redis and mqtt input in standalone and otel mode.

Checklist

• My code follows the style guidelines of this project
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation
• I have made corresponding change to the default configuration files
• I have added tests that prove my fix is effective or that my feature works. Where relevant, I have used the stresstest.sh script to run them under stress conditions and race detector to verify their stability.
• I have added an entry in ./changelog/fragments using the changelog tool.

Disruptive User Impact

None

How to test this PR locally

cd x-pack/filebeat
go test -run TestRedisInputOTelE2E ./tests/integration  -tags integration -v 
go test -run TestMQTTInputOTelE2E -tags integration ./tests/integration

Related issues

• Partially fixes https://github.com/elastic/ingest-dev/issues/4889

[github-actions]

🤖 GitHub comments

Just comment with:

• run docs-build : Re-trigger the docs validation. (use unformatted text in the comment!)
• /test : Run the Buildkite pipeline.

[mergify]

This pull request does not have a backport label.
If this is a bug or security fix, could you label this PR @khushijain21? 🙏.
For such, you'll need to label your PR with:

• The upcoming major version of the Elastic Stack
• The upcoming minor version of the Elastic Stack (if you're not pushing a breaking change)

To fixup this pull request, you need to add the backport labels for the needed
branches, such as:

• backport-8./d is the label to automatically backport to the 8./d branch. /d is the digit
• backport-active-all is the label that automatically backports to all active branches.
• backport-active-8 is the label that automatically backports to all active minor branches for the 8 major.
• backport-active-9 is the label that automatically backports to all active minor branches for the 9 major.

[infra-vault-gh-plugin-prod]

Pinging @elastic/elastic-agent-data-plane (Team:Elastic-Agent-Data-Plane)

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Enterprise

Run ID: 7f4533fe-bf94-48f7-9aa3-442354b81a4f

📥 Commits

Reviewing files that changed from the base of the PR and between f7cbabe and 51faf94.

📒 Files selected for processing (1)

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

🚧 Files skipped from review as they are similar to previous changes (1)

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

---

📝 Walkthrough

Walkthrough

The PR adds filebeat/input/redis/testutil with Redis host/port, client, slowlog, emission, and TLS certificate helpers, updates Redis integration tests and compose/image wiring for Redis TLS and shared test services, adds Filebeat-vs-OTel integration tests for CEL, HTTPJSON, Redis, and MQTT inputs, removes the older HTTPJSON integration test, and increases the OTel collector startup wait to 15 seconds.

Possibly related PRs

• elastic/beats#51091: Removes TestCELInputOTelE2E from x-pack/filebeat/tests/integration/otel_cel_test.go, which is the same CEL OTel E2E test added in this PR.

Suggested labels

enhancement, Filebeat

🚥 Pre-merge checks | ✅ 2

✅ Passed checks (2 passed)

Check name	Status	Explanation
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

• 🛠️ Update Documentation

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[coderabbitai]

Actionable comments posted: 1

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@filebeat/input/redis/testutil/testutil.go`:
- Around line 92-109: The emission loop in testutil’s goroutine uses
require.NoError for both conn.Close and conn.Do, which is unsafe because require
must not run in a background goroutine. Update the goroutine around conn.Get/Do
and ticker handling to send any error back to the main test via a channel or
shared error value, then perform the require/assertion in the parent test
function after the goroutine completes.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Enterprise

Run ID: fd9f2c9f-6add-4fef-b034-622547e13daf

📥 Commits

Reviewing files that changed from the base of the PR and between 0762870 and eac8fab.

📒 Files selected for processing (6)

• filebeat/input/redis/redis_integration_test.go
• filebeat/input/redis/testutil/testutil.go
• x-pack/filebeat/tests/integration/otel_cel_test.go
• x-pack/filebeat/tests/integration/otel_filebeat_input_test.go
• x-pack/filebeat/tests/integration/otel_test.go
• x-pack/otel/oteltestcol/collector.go

💤 Files with no reviewable changes (2)

• x-pack/filebeat/tests/integration/otel_cel_test.go
• x-pack/filebeat/tests/integration/otel_test.go

[coderabbitai]

Actionable comments posted: 1

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@x-pack/filebeat/docker-compose.yml`:
- Line 12: Add a healthcheck to the new redis service in the docker-compose
setup, or change proxy_dep so it no longer depends on redis with
service_healthy. The unsatisfied dependency is introduced by the redis entry and
the proxy_dep service relationship, so update whichever of those is intended to
coordinate startup. Keep the compose dependency consistent by ensuring the
referenced service actually exposes a health status before using
service_healthy.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Enterprise

Run ID: a1ca60e5-6769-4e8e-b520-da7e30d5ee76

📥 Commits

Reviewing files that changed from the base of the PR and between eac8fab and eee8fc0.

📒 Files selected for processing (1)

• x-pack/filebeat/docker-compose.yml

[coderabbitai]

Actionable comments posted: 1

♻️ Duplicate comments (1)

x-pack/filebeat/docker-compose.yml (1)

13-13: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

mosquitto is gated on service_healthy but no healthcheck is defined.

proxy_dep now waits for mosquitto to become healthy, but the new mosquitto service only exposes build and ports, and the supplied testing/environments/docker/mosquitto/Dockerfile:1-3 snippet does not define an image HEALTHCHECK. That makes this dependency unsatisfiable and can stall the integration stack before tests start.

Also applies to: 119-123

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@x-pack/filebeat/docker-compose.yml` at line 13, The mosquitto dependency in
the docker-compose setup is using service_healthy without any healthcheck to
satisfy it. Add a healthcheck for the mosquitto service itself (or change the
dependency in proxy_dep to a condition that matches the available startup
signal) so the compose graph can resolve readiness; use the mosquitto service
definition and proxy_dep dependency entries to locate the affected config, and
ensure the testing/environments/docker/mosquitto Dockerfile or compose service
provides the needed health status.

Source: Path instructions

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@filebeat/input/mqtt/testutil/testutil.go`:
- Around line 65-95: The publish assertions in PublishMessage and EmitMessages
are running inside a background goroutine, so failures won’t reliably fail the
test from the main goroutine. Refactor EmitMessages (and the related publisher
goroutine usage in mqtt_integration_test.go) to return publish errors or send
them back to the test goroutine, then perform the require/assert checks there;
keep the publishing loop in the goroutine but move the require.True/NoError/Fail
calls out of it.

---

Duplicate comments:
In `@x-pack/filebeat/docker-compose.yml`:
- Line 13: The mosquitto dependency in the docker-compose setup is using
service_healthy without any healthcheck to satisfy it. Add a healthcheck for the
mosquitto service itself (or change the dependency in proxy_dep to a condition
that matches the available startup signal) so the compose graph can resolve
readiness; use the mosquitto service definition and proxy_dep dependency entries
to locate the affected config, and ensure the
testing/environments/docker/mosquitto Dockerfile or compose service provides the
needed health status.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Enterprise

Run ID: 3bcf3255-d7d6-4404-af37-eda7bb67f60b

📥 Commits

Reviewing files that changed from the base of the PR and between a5def91 and f7cbabe.

📒 Files selected for processing (4)

• filebeat/input/mqtt/mqtt_integration_test.go
• filebeat/input/mqtt/testutil/testutil.go
• x-pack/filebeat/docker-compose.yml
• x-pack/filebeat/tests/integration/otel_filebeat_input_test.go

[belimawr]

This PR is quite large, the refactoring/moving tests around could have been done in a different PR to make reviewing easier. Could you split it into two PRs?

[khushijain21]

This PR is quite large, the refactoring/moving tests around could have been done in a different PR to make reviewing easier. Could you split it into two PRs?

Done #51699

[mergify]

This pull request is now in conflicts. Could you fix it? 🙏
To fixup this pull request, you can check out it locally. See documentation: https://help.github.com/articles/checking-out-pull-requests-locally/

git fetch upstream
git checkout -b redis-input-test upstream/redis-input-test
git merge upstream/main
git push upstream redis-input-test

[github-actions]

TL;DR

Both failed x-pack Filebeat integration jobs are failing at Go test compile time because deleteDataStreamsFromES is now declared twice in the integration package. Remove the duplicate helper from the newly added otel_filebeat_input_test.go and reuse the existing helper from otel_kafka_test.go.

Remediation

• Delete deleteDataStreamsFromES from x-pack/filebeat/tests/integration/otel_filebeat_input_test.go:855-858 and remove the now-unused github.com/elastic/go-elasticsearch/v8 import from that file.
• Re-run cd x-pack/filebeat && go test -c -tags integration ./tests/integration or the Buildkite targets mage goIntegTest and mage goFIPSOnlyIntegTest to confirm the package compiles.

Investigation details

Root Cause

This is a code/test compile failure introduced by adding a second package-level helper with the same name. The PR-head file x-pack/filebeat/tests/integration/otel_filebeat_input_test.go:855-858 adds:

func deleteDataStreamsFromES(t *testing.T, es *elasticsearch.Client, dataStreams []string) {
	_, err := es.Indices.DeleteDataStream(dataStreams)
	require.NoError(t, err, "failed to delete data streams")
}

But x-pack/filebeat/tests/integration/otel_kafka_test.go:436-439 already defines deleteDataStreamsFromES in the same package integration, so the integration test package cannot build. The new tests can call the existing helper directly once the duplicate definition is removed.

Evidence

• Build: https://buildkite.com/elastic/beats/builds/49007
• Failed jobs: :ubuntu: x-pack/filebeat: Go Integration Tests and :ubuntu: x-pack/filebeat: Go fips140=only Integration Tests
• Key log excerpt from both jobs:

# github.com/elastic/beats/v7/x-pack/filebeat/tests/integration [github.com/elastic/beats/v7/x-pack/filebeat/tests/integration.test]
tests/integration/otel_kafka_test.go:436:6: deleteDataStreamsFromES redeclared in this block
	tests/integration/otel_filebeat_input_test.go:855:6: other declaration of deleteDataStreamsFromES
FAIL	github.com/elastic/beats/v7/x-pack/filebeat/tests/integration [build failed]

Verification

Not run locally: the checkout available to this workflow is main, while the failure is on PR head 7a356bdde510a0683daec22615f7ec5579d5a8c2. Static verification against the PR-head file contents confirmed both declarations and the unused import that remains after removing the duplicate helper.

---

What is this? | From workflow: PR Buildkite Detective

Give us feedback! React with 🚀 if perfect, 👍 if helpful, 👎 if not.