Use receivertest package to test beats receivers

[mauri870]

The open telemetry project has a receivertest module that can be used to test a receiver's contract. I'm particularly interested in the CheckConsumerContract function that is used to test the contract between the receiver and the next consumer in the pipeline.

This test has a couple of interesting scenarios such as:

• Always succeed. We expect all data to be delivered as is.
• Random non-permanent error
• Random permanent error
• Random error

This test is based on a Generator that is responsible to produce the data used during the test. For this to work properly the beats receiver must implement the ReceiveLogs(data plog.Logs) function to receive the logs from this external source. In the case of beats receivers, the beats themselves produce their own data, so we need to come up with a way to adapt this generator concept or implement said function. I believe the messages should be unique, so perhaps the benchmark input being able to output unique strings or an integer counter should be enough.

This seems quite handy to setup once and use the same machinery to test every beats receiver.

Here is an example test:

// TestConsumeContract is an example of testing of the receiver for the contract between the
// receiver and next consumer.
func TestConsumeContract(t *testing.T) {
	// Number of log records to send per scenario.
	const logsPerTest = 100

	generator := &exampleLogGenerator{t: t}
	cfg := &exampleReceiverConfig{generator: generator}

	params := CheckConsumeContractParams{
		T:             t,
		Factory:       newExampleFactory(),
		Signal:        pipeline.SignalLogs,
		Config:        cfg,
		Generator:     generator,
		GenerateCount: logsPerTest,
	}

	// Run the contract checker. This will trigger test failures if any problems are found.
	CheckConsumeContract(params)
}

During my initial experimentation with this I found a bug with global state in libbeat that is now resolved by #41475.

[elasticmachine]

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

[mauri870]

I investigated trying to use the receivertest for tests and have some findings.

The two main challenges to use this package is to write a generator implementation that creates unique events that can be processed by the receiver and tell receivertest where it should look for this id in the plog.Logs.

For the first problem, the way I found to adapt this generator concept to Beats receiver is to have a generator that writes ndjson lines to a file with a specific id and use the file as a filestream input with ndjson parser and a document_id option. This seems to work properly during testing.

Regarding the second challenge, it is a bit more complicated. Unfortunately we cannot use the receivertest as-is, it looks for a hardcodedtest_id attribute in the LogRecord to check the delivery of a specific event created by the generator, but in beats the otelconsumer writes everything to the LogRecord body.

We would have to extend receivertest to have the ability to use a user defined function to lookup this ID. I have a prototype ready at main...mauri870:beats:receivertest. I have opened open-telemetry/opentelemetry-collector#12003 upstream as well.