[beats receivers] Respect Filebeat's ability to request guaranteed send on a per event basis with infinite retries

[cmacknz]

Filebeat by default retries indefinitely, while the other Beats default to a maximum of 3 retries. Generated exporter configurations need a way to remain compatible with this.

Filebeat https://www.elastic.co/docs/reference/beats/filebeat/elasticsearch-output#_max_retries

Filebeat ignores the max_retries setting and retries indefinitely.

Metricbeat https://www.elastic.co/docs/reference/beats/metricbeat/elasticsearch-output#_max_retries

The default is 3.

Packetbeat https://www.elastic.co/docs/reference/beats/packetbeat/elasticsearch-output#_max_retries

The default is 3.

Auditbeat https://www.elastic.co/docs/reference/beats/auditbeat/elasticsearch-output#_max_retries

The default is 3.

Heartbeat https://www.elastic.co/docs/reference/beats/heartbeat/elasticsearch-output#_max_retries

The default is 3.

The indefinite retries in Filebeat are configured as part of pipeline setup seprate from the output

beats/filebeat/beater/filebeat.go

Lines 359 to 362 in 09693ac

	// Filebeat by default required infinite retry. Let's configure this for all
	// inputs by default. Inputs (and InputController) can overwrite the sending
	// guarantees explicitly when connecting with the pipeline.
	fb.pipeline = pipetool.WithDefaultGuarantees(fb.pipeline, beat.GuaranteedSend)

It also appears some specific inputs may deviate from this, for example netflow

beats/x-pack/filebeat/input/netflow/input.go

Lines 155 to 161 in 09693ac

	client, err := connector.ConnectWith(beat.ClientConfig{
	PublishMode: beat.DefaultGuarantees,
	Processing: beat.ProcessingConfig{
	EventNormalization: boolPtr(false),
	},
	EventListener: nil,
	})

The Beats pipeline supports per event retry policies right now which is what allows this:

beats/libbeat/publisher/pipeline/pipeline.go

Lines 197 to 202 in 09693ac

	switch cfg.PublishMode {
	case beat.GuaranteedSend:
	eventFlags = publisher.GuaranteedSend
	case beat.DropIfFull:
	canDrop = true
	}

beats/libbeat/publisher/pipeline/ttl_batch.go

Lines 181 to 187 in 09693ac

	// filter for events with guaranteed send flags
	events := b.events[:0]
	for _, event := range b.events {
	if event.Guaranteed() {
	events = append(events, event)
	}
	}

[elasticmachine]

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