[Elastic Agent] Prototype a shipper gRPC input for Filebeat

[cmacknz]

Create a prototype input for Filebeat that implements the shipper API. The intent is to implement the data shipper as another instance of Filebeat so that we can completely reuse the existing Beat event pipeline and outputs.

The shipper input will be Elastic licensed and should be added to the https://github.com/elastic/beats/tree/main/x-pack/filebeat/input directory.

The shipper input should ideally start a single gRPC server and create a new beat.Client for each unique data stream. This will allow each data stream to have its own processor configuration and allows for shared processor pipelines to execute concurrently. This follows the pattern used in the AWS S3 input in https://github.com/elastic/beats/pull/33658.

We can explore other approaches if the ideal approach is initially too complex. The path of least resistance will likely be to have one instance of the shipper input for each input unit received from the Elastic Agent, where there is an input unit per component connected to the shipper. This will result in multiple gRPC server instances, but given that the shipper gRPC communication uses Unix domain sockets / named pipes this may be an acceptable overhead.

Acceptance Criteria

• [ ] An integration test exists proving that a Filebeat using the shipper output can publish events through a Filebeat that is running the new shipper input.

[elasticmachine]

Pinging @elastic/elastic-agent (Team:Elastic-Agent)

[cmacknz]

The ideal outcome here is that an system or integration test exists proving that a Filebeat instance running the shipper output can communicate with a shipper instance running the shipper input.

[fearful-symmetry]

So, currently tinkering with filebeat to see how we can do this. There's a few open questions:

• How do we handle multiple inputs, considering that the beat central management client will reload runners that have updated configs? Should each shipper input be a separate filebeat input instance, with its own gRPC server?
• How should config be structured? With the old shipper we had discrete units, however we a filebeat input we get a single config object. Should we just expect a struct with a section for gRPC connection data, and another field for the input data?

[cmacknz]

This is where the agent create's the shipper input configurations now: https://github.com/elastic/elastic-agent/blob/eaa9d0ac42986259e153482170e077ea8e191c5a/pkg/component/runtime/manager_shipper.go#L37-L51

[cmacknz]

Spoke today about how the design should work, I've updated the diagram in the description to match.

Summary of the implementation:

• There will be one instance of the shipper protocol input per component. This allows us to report status back to the agent about the health of the input for each component connection and per-input metrics automatically map back to components.
• Each shipper input will have its own gRPC server with a unique Unix socket connection per component. This will require us to update the agent to create a socket per component. It currently creates a single socket shared by all components. We will defer this as out of scope in this issue and demo with only a single connected component.
• The agent must set the type parameter in the input units it sends to the shipper to match the name of the shipper protocol input. This allows Filebeat to create and manage the lifecycle of the shipper inputs with no other modifications to the shipper. This is the code in the agent that needs to be modified.
• Each shipper input will create a beat.Client for each data stream. Each input receives a list of streams configured for a component along with the processors to apply to that stream. The input must be a map from data stream name to beat.Client to ensure the correct processors are applied to each event.

Alternatives considered:

• A single gRPC server and shipper input. This would require changing the internals of the agent to provision a single input in the shipper which is too much work for the prototype. It would also not allow us to easily report per component metrics and status information.
• Multiple inputs with a shared gRPC server. Discarded because we would need a way to associate events received on the server to the correct input to process them which is unnecessarily complex.

[blakerouse]

• The agent must set the type parameter in the input units it sends to the shipper to match the name of the shipper protocol input. This allows Filebeat to create and manage the lifecycle of the shipper inputs with no other modifications to the shipper. This is the code in the agent that needs to be modified.

This is actually fixed in this PR - https://github.com/elastic/elastic-agent/pull/2543

[cmacknz]

Great, then we just need to change the shipper spec file to have the name parameter match the input type we plan to run:

https://github.com/elastic/elastic-agent/blob/05b4c010ab20ef18a0f857a2de0eccb0bfac5d01/pkg/component/load.go#L174-L175

            shipperSpecs[shipper.Name] = ShipperRuntimeSpec{
                ShipperType: shipper.Name

https://github.com/elastic/elastic-agent-shipper/blob/70cd54ce1b7fa041d1fdec62335137ef08136d4b/elastic-agent-shipper.spec.yml#L3

shippers:
  - name: shipper
    description: "Elastic Agent Shipper"
    platforms: