[aws] feat: aws-s3 input registry cleanup for untracked s3 objects

[Kavindu-Dodan]

Proposed commit message

This PR partially addresses #39116 by introducing a registry cleanup strategy for aws-s3 input.

The cleanup implemented here removes registry entries if the s3 object is no longer available (aka tracked) when listing inside the polling lookup. The cleanup removes objects that are not tracked from both the local state and internal store (backed by the registry) to reduce the memory usage.

Note that, this only benefits when s3 objects get removed (ex:- using lifecycle policy) and are no longer available. There should be a follow-up for instances where such removal is not done at the bucket. For example, this could be done by,

• Defining a time-based sliding window (ex:- only consider s3 objects of the past 3 days)
• Only accept objects of a known storage class (ex:- S3 Standard ) so that users can archive objects ¹

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
• I have added an entry in CHANGELOG.next.asciidoc or CHANGELOG-developer.next.asciidoc.

How to test this PR locally

• Build or run filebeat with aws-s3 input
• Add s3 objects and observe memory growth with pprof (regex filter for awss3)
• Remove s3 objects and observe memory reduction

Screenshots

Given below are pprof analyss comparisons for ~4K objects in registry and once they were clean up by removing S3 objects (emptying the bucket)

• With ~4K objects processed

• When S3 objects were removed

Related issues

• Relates #39116
• Depends on [AWS] improve S3 input states copy by only storing filtered entries #41869

Footnotes

1. https://docs.aws.amazon.com/AmazonS3/latest/userguide/lifecycle-transition-general-considerations.html ↩

[elasticmachine]

Pinging @elastic/obs-ds-hosted-services (Team:obs-ds-hosted-services)

[kaiyan-sheng]

Nice work @Kavindu-Dodan ! One quick question first, should we also introduce clean_removed parameter just to be matching the filestream input?

[Kavindu-Dodan]

Nice work @Kavindu-Dodan ! One quick question first, should we also introduce clean_removed parameter just to be matching the filestream input?

For s3 input, if clean_removed is set to false, then we will avoid cleanup of registry even for s3 objects that are no longer visible when listing through ListObjectsV2 API. This again can overload the in-memory state store. Besides, there's a core difference between filestream input and s3. In filestream input, we can continuously get new events as file get updated. Whereas in s3, we mark object processed as soon as we process and get ack for all events generated after processing it. So, personally I do not think we need to introduce clean_removed for s3 input unless we see another usage for this. Let me know your view.

[kaiyan-sheng]

Looks good to me!

[elasticmachine]

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

[lucabelluccini]

Looking at the change, if I've correctly understood, we're removing the IDs from the registry once the files have been read.
Shouldn't be removing the entries only when we've acknowledged the shipping of the logs downstream (to Elasticsearch/Outputs)?

[Kavindu-Dodan]

Looking at the change, if I've correctly understood, we're removing the IDs from the registry once the files have been read. Shouldn't be removing the entries only when we've acknowledged the shipping of the logs downstream (to Elasticsearch/Outputs)?

The cleanup happens for objects we no longer see in the configured bucket and they won't be processed so there won't be any elastic acknowledgment. Any known bucket entry will be kept in the registry and will be tracked for their success or failure. See this logic [1] which is unchanged.

[1] -

beats/x-pack/filebeat/input/awss3/s3_input.go

Lines 185 to 193 in 9b29500

	acks.Add(publishCount, func() {
	err := in.states.AddState(state)
	if err != nil {
	in.log.Errorf("saving completed object state: %v", err.Error())
	}
	// Metrics
	in.metrics.s3ObjectsAckedTotal.Inc()
	})