Inject Beat gRPC stream identifiers without relying on agent specification file transformations

[cmacknz]

To quickly allow metricbeat and filebeat to integrate with the shipper the stream identifiers (data stream and input ID) needed to allow the shipper to map events to processors are being injected via the inject_stream_processor specification file transformation. See the implementation in elastic/elastic-agent#527.

In the V2 agent specification file format the agent will no longer be able automatically transform sections of the agent policy into a beat configuration file. All inputs must know how to configure themselves to run under agent automatically.

Define and implement a way of providing the needed stream identifiers to agent inputs without relying on specification file transformations as required by the V2 specification file format. This includes all beats (not just filebeat and metricbeat) and non-beat agent inputs like endpoint security.

Ideally the beats can consume the agent policy sections containing the relevant information directly without transformation, and instantiate the necessary processors automatically. This is similar to the approach heartbeat currently uses, see the heartbeat specification file and heartbeat code adding the processors.

Relates To:

• [Meta] Remove Elastic Agent V1 control protocol elastic-agent-client#32
• [Meta][Feature] The agent policy's input processor configurations must be provided to the shipper. elastic-agent#220
  • Provide the relevant subset of the agent policy to the shipper as its configuration elastic-agent#617

This task is considered complete when:

1. A test exists proving that the shipper can map processors to events sent from filebeat and metricbeat without using any configuration file transformations.
2. The implementation used for metricbeat and filebeat is available for reuse in libbeat or elastic-agent-libs, such that any existing beat can easily be updated to add the new stream identifiers to produced events when communicating with the shipper.

[cmacknz]

@fearful-symmetry I believe the work remaining here is to use a config transformation to automatically add the metadata add_fields processors from elastic/elastic-agent#527 to each of the Beats on the v2 feature branch.

[fearful-symmetry]

@cmacknz A little confused here. The processors should all be added on the beats side, are you talking about importing them into the shipper?

[cmacknz]

This issue was created before we had any of the v2 implementation in Beats. You are correct that the remaining work should happen in Beats, this issue will be transferred to the Beats repository.

[cmacknz]

Synced up with Alex, this has already been taken care of. Closing.

beats/x-pack/libbeat/management/generate.go

Lines 173 to 176 in a528bd2

	if expected.GetId() != "" {
	inputId := generateAddFieldsProcessor(mapstr.M{"input_id": expected.Id}, "@metadata")
	processors = append(processors, inputId)
	}

beats/x-pack/libbeat/management/generate.go

Lines 194 to 195 in a528bd2

	streamID := streamExpected.GetId()
	sourceStream := generateAddFieldsProcessor(mapstr.M{"stream_id": streamID}, "@metadata")