Skip to content

Commit d17c140

Browse files
kcreddykcreddy
authored
x-pack/filebeat/input/awss3: Add lexicographical polling mode (#48310)
Adds lexicographical ordering mode for S3 bucket polling. The input uses the S3 "StartAfter" parameter to resume listing. For buckets with a large number of objects where object keys are naturally ordered (ex: timestamp-prefixed), this significantly reduces the number of objects listed [1] and memory footprint. The persisted tail key tracks in-flight objects for crash resistance, ensuring no objects are skipped after an unexpected restart. Adds 2 new configuration options "lexicographical_ordering" and "lexicographical_lookback_keys": - lexicographical_ordering: When set to "true", enables lexicographical ordering mode. Default is "false". - lexicographical_lookback_keys: Maintains a bounded lookback buffer of the N most recently processed keys (default: 100). The start-after parameter uses the lexicographically oldest key in the buffer, creating a sliding window that catches late-arriving objects. [1] #47926 (comment)
1 parent 22bedbc commit d17c140

20 files changed

Lines changed: 1954 additions & 236 deletions

‎changelog/fragments/1767727700-lexicographical-polling-mode-to-AWS-S3.yaml‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
# Kind can be one of:
2+
# - breaking-change: a change to previously-documented behavior
3+
# - deprecation: functionality that is being removed in a later release
4+
# - bug-fix: fixes a problem in a previous version
5+
# - enhancement: extends functionality but does not break or fix existing behavior
6+
# - feature: new functionality
7+
# - known-issue: problems that we are aware of in a given version
8+
# - security: impacts on the security of a product or a user’s deployment.
9+
# - upgrade: important information for someone upgrading from a prior version
10+
# - other: does not fit into any of the other categories
11+
kind: feature
12+
13+
# Change summary; a 80ish characters long description of the change.
14+
summary: Add lexicographical polling mode to AWS-S3 input
15+
16+
# Long description; in case the summary is not enough to describe the change
17+
# this field accommodate a description without length limits.
18+
# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
19+
#description:
20+
21+
# Affected component; a word indicating the component this changeset affects.
22+
component: filebeat
23+
24+
# PR URL; optional; the PR number that added the changeset.
25+
# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
26+
# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
27+
# Please provide it if you are adding a fragment for a different PR.
28+
#pr: https://github.com/owner/repo/1234
29+
30+
# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
31+
# If not present is automatically filled by the tooling with the issue linked to the PR number.
32+
#issue: https://github.com/owner/repo/1234
33+
issue: https://github.com/elastic/beats/issues/47926

‎docs/reference/filebeat/filebeat-input-aws-s3.md‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -453,6 +453,40 @@ Time interval for polling listing of the S3 bucket: default to `120s`.
453453
Prefix to apply for the list request to the S3 bucket. Default empty.
454454

455455

456+
### `lexicographical_ordering` [_lexicographical_ordering]
457+
458+
```{applies_to}
459+
stack: ga 9.4.0
460+
```
461+
462+
When set to `true`, enables lexicographical ordering mode for S3 bucket polling. In this mode, the input uses the S3 `StartAfter` parameter to resume listing from the last processed object key, which can significantly reduce the number of objects listed in each polling cycle. This is particularly useful for buckets with many objects where object keys are naturally ordered (for example, timestamp-prefixed keys such as `CloudTrail/us-east-1/2025/01/15/log-001.json.gz`).
463+
464+
When enabled, the input maintains a state of processed object keys and uses the key which is _lexicographically least_ as the starting point for subsequent list operations. The number of keys tracked is controlled by `lexicographical_lookback_keys`.
465+
466+
Default: `false`
467+
468+
::::{note}
469+
This option is only applicable when using S3 bucket polling (`bucket_arn` or `non_aws_bucket_name`). It has no effect when using SQS notifications. Using lexicographical ordering mode assumes that objects written to the bucket are immutable, and any changes to the object remains undetected.
470+
::::
471+
472+
473+
### `lexicographical_lookback_keys` [_lexicographical_lookback_keys]
474+
475+
```{applies_to}
476+
stack: ga 9.4.0
477+
```
478+
479+
Specifies the maximum number of S3 object keys to track in memory when `lexicographical_ordering` is enabled. This value determines how many recently processed object keys are retained to support the `StartAfter` functionality. The lookback buffer ensures objects that arrive out of lexicographical order (due to late delivery, clock skew, or retries) are still detected and processed.
480+
481+
A higher value provides more resilience against reprocessing objects when new objects with older keys arrive, but consumes more memory. A lower value reduces memory usage but can cause more objects to be re-listed if objects with older keys are added to the bucket.
482+
483+
Default: `100`
484+
485+
::::{note}
486+
This option only takes effect when `lexicographical_ordering` is set to `true`.
487+
::::
488+
489+
456490
### `number_of_workers` [_number_of_workers_2]
457491

458492
Number of workers that will process the S3 or SQS objects listed. Required when `bucket_arn` or `access_point_arn` is set, otherwise (in the SQS case) defaults to 5.
@@ -1152,5 +1186,8 @@ This input exposes metrics under the [HTTP monitoring endpoint](/reference/fileb
11521186
| `s3_events_created_total` | Number of events created from processing S3 data. |
11531187
| `s3_objects_inflight_gauge` | Number of S3 objects inflight (gauge). |
11541188
| `s3_object_processing_time` | Histogram of the elapsed S3 object processing times in nanoseconds (start of download to completion of parsing). |
1189+
| `s3_polling_run_time` | Histogram of the elapsed time for each S3 polling run in nanoseconds. Only applicable when using S3 bucket polling. |
1190+
| `s3_polling_run_time_total` | Cumulative time spent in S3 polling runs in nanoseconds. Only applicable when using S3 bucket polling. |
1191+
| `s3_objects_listed_per_run` | Histogram of the number of S3 objects listed in each polling run. Only applicable when using S3 bucket polling. |
11551192

11561193

‎x-pack/filebeat/input/awss3/config.go‎

Lines changed: 43 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -27,41 +27,45 @@ import (
2727
)
2828

2929
type config struct {
30-
APITimeout time.Duration `config:"api_timeout"`
31-
AWSConfig awscommon.ConfigAWS `config:",inline"`
32-
AccessPointARN string `config:"access_point_arn"`
33-
BackupConfig backupConfig `config:",inline"`
34-
BucketARN string `config:"bucket_arn"`
35-
BucketListInterval time.Duration `config:"bucket_list_interval"`
36-
BucketListPrefix string `config:"bucket_list_prefix"`
37-
FileSelectors []fileSelectorConfig `config:"file_selectors"`
38-
IgnoreOlder time.Duration `config:"ignore_older"`
39-
NonAWSBucketName string `config:"non_aws_bucket_name"`
40-
NumberOfWorkers int `config:"number_of_workers"`
41-
PathStyle bool `config:"path_style"`
42-
ProviderOverride string `config:"provider"`
43-
QueueURL string `config:"queue_url"`
44-
ReaderConfig readerConfig `config:",inline"` // Reader options to apply when no file_selectors are used.
45-
RegionName string `config:"region"`
46-
SQSMaxReceiveCount int `config:"sqs.max_receive_count"` // The max number of times a message should be received (retried) before deleting it.
47-
SQSScript *scriptConfig `config:"sqs.notification_parsing_script"`
48-
SQSWaitTime time.Duration `config:"sqs.wait_time"` // The max duration for which the SQS ReceiveMessage call waits for a message to arrive in the queue before returning.
49-
SQSGraceTime time.Duration `config:"sqs.shutdown_grace_time"` // The time that the processing loop will wait for messages before shutting down.
50-
StartTimestamp string `config:"start_timestamp"`
51-
VisibilityTimeout time.Duration `config:"visibility_timeout"`
30+
APITimeout time.Duration `config:"api_timeout"`
31+
AWSConfig awscommon.ConfigAWS `config:",inline"`
32+
AccessPointARN string `config:"access_point_arn"`
33+
BackupConfig backupConfig `config:",inline"`
34+
BucketARN string `config:"bucket_arn"`
35+
BucketListInterval time.Duration `config:"bucket_list_interval"`
36+
BucketListPrefix string `config:"bucket_list_prefix"`
37+
FileSelectors []fileSelectorConfig `config:"file_selectors"`
38+
IgnoreOlder time.Duration `config:"ignore_older"`
39+
LexicographicalOrdering bool `config:"lexicographical_ordering"`
40+
LexicographicalLookbackKeys int `config:"lexicographical_lookback_keys"`
41+
NonAWSBucketName string `config:"non_aws_bucket_name"`
42+
NumberOfWorkers int `config:"number_of_workers"`
43+
PathStyle bool `config:"path_style"`
44+
ProviderOverride string `config:"provider"`
45+
QueueURL string `config:"queue_url"`
46+
ReaderConfig readerConfig `config:",inline"` // Reader options to apply when no file_selectors are used.
47+
RegionName string `config:"region"`
48+
SQSMaxReceiveCount int `config:"sqs.max_receive_count"` // The max number of times a message should be received (retried) before deleting it.
49+
SQSScript *scriptConfig `config:"sqs.notification_parsing_script"`
50+
SQSWaitTime time.Duration `config:"sqs.wait_time"` // The max duration for which the SQS ReceiveMessage call waits for a message to arrive in the queue before returning.
51+
SQSGraceTime time.Duration `config:"sqs.shutdown_grace_time"` // The time that the processing loop will wait for messages before shutting down.
52+
StartTimestamp string `config:"start_timestamp"`
53+
VisibilityTimeout time.Duration `config:"visibility_timeout"`
5254
}
5355

5456
func defaultConfig() config {
5557
c := config{
56-
APITimeout: 120 * time.Second,
57-
VisibilityTimeout: 300 * time.Second,
58-
BucketListInterval: 120 * time.Second,
59-
BucketListPrefix: "",
60-
SQSWaitTime: 20 * time.Second,
61-
SQSGraceTime: 20 * time.Second,
62-
SQSMaxReceiveCount: 5,
63-
NumberOfWorkers: 5,
64-
PathStyle: false,
58+
APITimeout: 120 * time.Second,
59+
VisibilityTimeout: 300 * time.Second,
60+
BucketListInterval: 120 * time.Second,
61+
BucketListPrefix: "",
62+
LexicographicalOrdering: false,
63+
LexicographicalLookbackKeys: 100,
64+
SQSWaitTime: 20 * time.Second,
65+
SQSGraceTime: 20 * time.Second,
66+
SQSMaxReceiveCount: 5,
67+
NumberOfWorkers: 5,
68+
PathStyle: false,
6569
}
6670
c.ReaderConfig.InitDefaults()
6771
return c
@@ -94,6 +98,14 @@ func (c *config) Validate() error {
9498
return fmt.Errorf("invalid format for access_point_arn <%v>", c.AccessPointARN)
9599
}
96100

101+
if c.LexicographicalOrdering && c.BucketARN == "" && c.AccessPointARN == "" && c.NonAWSBucketName == "" {
102+
return errors.New("lexicographical_ordering can only be used when polling AWS S3, S3 Access Point, or non-AWS S3 bucket")
103+
}
104+
105+
if c.LexicographicalOrdering && c.LexicographicalLookbackKeys <= 0 {
106+
return fmt.Errorf("lexicographical_lookback_keys <%d> must be greater than 0", c.LexicographicalLookbackKeys)
107+
}
108+
97109
if c.QueueURL != "" && (c.VisibilityTimeout <= 0 || c.VisibilityTimeout.Hours() > 12) {
98110
return fmt.Errorf("visibility_timeout <%v> must be greater than 0 and "+
99111
"less than or equal to 12h", c.VisibilityTimeout)

‎x-pack/filebeat/input/awss3/config_test.go‎

Lines changed: 14 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -31,19 +31,20 @@ func TestConfig(t *testing.T) {
3131
parserConf := parser.Config{}
3232
require.NoError(t, parserConf.Unpack(conf.MustNewConfigFrom("")))
3333
return config{
34-
QueueURL: quequeURL,
35-
BucketARN: s3Bucket,
36-
AccessPointARN: s3AccessPoint,
37-
NonAWSBucketName: nonAWSS3Bucket,
38-
APITimeout: 120 * time.Second,
39-
VisibilityTimeout: 300 * time.Second,
40-
SQSMaxReceiveCount: 5,
41-
SQSWaitTime: 20 * time.Second,
42-
SQSGraceTime: 20 * time.Second,
43-
BucketListInterval: 120 * time.Second,
44-
BucketListPrefix: "",
45-
PathStyle: false,
46-
NumberOfWorkers: 5,
34+
QueueURL: quequeURL,
35+
BucketARN: s3Bucket,
36+
AccessPointARN: s3AccessPoint,
37+
NonAWSBucketName: nonAWSS3Bucket,
38+
APITimeout: 120 * time.Second,
39+
VisibilityTimeout: 300 * time.Second,
40+
SQSMaxReceiveCount: 5,
41+
SQSWaitTime: 20 * time.Second,
42+
SQSGraceTime: 20 * time.Second,
43+
BucketListInterval: 120 * time.Second,
44+
BucketListPrefix: "",
45+
LexicographicalLookbackKeys: 100,
46+
PathStyle: false,
47+
NumberOfWorkers: 5,
4748
ReaderConfig: readerConfig{
4849
BufferSize: 16 * humanize.KiByte,
4950
MaxBytes: 10 * humanize.MiByte,

‎x-pack/filebeat/input/awss3/input_benchmark_test.go‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -157,7 +157,7 @@ func (c constantS3) DeleteObject(context.Context, string, string, string) (*s3.D
157157
return nil, nil
158158
}
159159

160-
func (c constantS3) ListObjectsPaginator(string, string) s3Pager {
160+
func (c constantS3) ListObjectsPaginator(_, _, _ string) s3Pager {
161161
return c.pagerConstant
162162
}
163163

@@ -342,20 +342,21 @@ func benchmarkInputS3(t *testing.T, numberOfWorkers int) testing.BenchmarkResult
342342
s3API.pagerConstant = newS3PagerConstant(curConfig.BucketListPrefix)
343343
store := openTestStatestore()
344344

345-
states, err := newStates(nil, store, "")
346-
assert.NoError(t, err, "states creation should succeed")
345+
registry, err := newStateRegistry(nil, store, "", false, 0)
346+
assert.NoError(t, err, "registry creation should succeed")
347347

348348
s3EventHandlerFactory := newS3ObjectProcessorFactory(metrics, s3API, config.FileSelectors, backupConfig{}, logp.NewNopLogger())
349349
s3Poller := &s3PollerInput{
350-
log: logp.NewLogger(inputName),
350+
log: log,
351351
config: config,
352352
metrics: metrics,
353353
s3: s3API,
354354
pipeline: pipeline,
355355
s3ObjectHandler: s3EventHandlerFactory,
356-
states: states,
356+
registry: registry,
357357
provider: "provider",
358358
filterProvider: newFilterProvider(&config),
359+
strategy: newPollingStrategy(config.LexicographicalOrdering, log),
359360
status: &statusReporterHelperMock{},
360361
}
361362

‎x-pack/filebeat/input/awss3/input_integration_test.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -726,10 +726,11 @@ func TestPaginatorListPrefix(t *testing.T) {
726726

727727
s3API := &awsS3API{
728728
client: s3Client,
729+
log: logp.NewNopLogger(),
729730
}
730731

731732
var objects []string
732-
paginator := s3API.ListObjectsPaginator(tfConfig.BucketName, "log")
733+
paginator := s3API.ListObjectsPaginator(tfConfig.BucketName, "log", "")
733734
for paginator.HasMorePages() {
734735
page, err := paginator.NextPage(context.Background())
735736
assert.NoError(t, err)

‎x-pack/filebeat/input/awss3/interfaces.go‎

Lines changed: 14 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,7 @@ type s3Mover interface {
7373
}
7474

7575
type s3Lister interface {
76-
ListObjectsPaginator(bucket, prefix string) s3Pager
76+
ListObjectsPaginator(bucket, prefix, startAfterKey string) s3Pager
7777
}
7878

7979
type s3Pager interface {
@@ -118,7 +118,7 @@ func (a *awsSQSAPI) ReceiveMessage(ctx context.Context, maxMessages int) ([]type
118118

119119
receiveMessageOutput, err := a.client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
120120
QueueUrl: awssdk.String(a.queueURL),
121-
MaxNumberOfMessages: int32(min(maxMessages, sqsMaxNumberOfMessagesLimit)),
121+
MaxNumberOfMessages: int32(min(maxMessages, sqsMaxNumberOfMessagesLimit)), //nolint:gosec // value is bounded by sqsMaxNumberOfMessagesLimit (10)
122122
VisibilityTimeout: int32(a.visibilityTimeout.Seconds()),
123123
WaitTimeSeconds: int32(a.longPollWaitTime.Seconds()),
124124
AttributeNames: []types.QueueAttributeName{sqsApproximateReceiveCountAttribute, sqsSentTimestampAttribute},
@@ -205,6 +205,7 @@ func (a *awsSQSAPI) GetQueueAttributes(ctx context.Context, attr []types.QueueAt
205205

206206
type awsS3API struct {
207207
client *s3.Client
208+
log *logp.Logger
208209

209210
// others is the set of other clients referred
210211
// to by notifications seen by the API connection.
@@ -216,8 +217,8 @@ type awsS3API struct {
216217

217218
const awsS3APIcacheMax = 100
218219

219-
func newAWSs3API(cli *s3.Client) *awsS3API {
220-
return &awsS3API{client: cli, others: make(map[string]*s3.Client)}
220+
func newAWSs3API(cli *s3.Client, log *logp.Logger) *awsS3API {
221+
return &awsS3API{client: cli, log: log, others: make(map[string]*s3.Client)}
221222
}
222223

223224
func (a *awsS3API) GetObject(ctx context.Context, region, bucket, key string) (*s3.GetObjectOutput, error) {
@@ -232,7 +233,7 @@ func (a *awsS3API) GetObject(ctx context.Context, region, bucket, key string) (*
232233
out middleware.FinalizeOutput, metadata middleware.Metadata, err error,
233234
) {
234235
out, metadata, err = next.HandleFinalize(ctx, in)
235-
requestURL, parseErr := url.Parse(in.Request.(*smithyhttp.Request).URL.String())
236+
requestURL, parseErr := url.Parse(in.Request.(*smithyhttp.Request).URL.String()) //nolint:errcheck // type assertion is guaranteed by AWS SDK
236237
if parseErr != nil {
237238
return out, metadata, err
238239
}
@@ -319,11 +320,16 @@ func (a *awsS3API) clientFor(region string) *s3.Client {
319320
return cli
320321
}
321322

322-
func (a *awsS3API) ListObjectsPaginator(bucket, prefix string) s3Pager {
323-
pager := s3.NewListObjectsV2Paginator(a.client, &s3.ListObjectsV2Input{
323+
func (a *awsS3API) ListObjectsPaginator(bucket, prefix, startAfterKey string) s3Pager {
324+
input := &s3.ListObjectsV2Input{
324325
Bucket: awssdk.String(bucket),
325326
Prefix: awssdk.String(prefix),
326-
})
327+
}
328+
if startAfterKey != "" {
329+
input.StartAfter = awssdk.String(startAfterKey)
330+
a.log.Debugw("Listing objects with startAfterKey.", "start_after_key", startAfterKey)
331+
}
332+
pager := s3.NewListObjectsV2Paginator(a.client, input)
327333

328334
return pager
329335
}

‎x-pack/filebeat/input/awss3/interfaces_test.go‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,14 +8,16 @@ import (
88
"testing"
99

1010
"github.com/aws/aws-sdk-go-v2/service/s3"
11+
12+
"github.com/elastic/elastic-agent-libs/logp"
1113
)
1214

1315
func TestAWSS3API_clientFor(t *testing.T) {
1416
// When SQS notifications do not contain a region (like Crowdstrike FDR's
1517
// custom notifications), then the default pre-made S3 client should be used.
1618
t.Run("empty_region_uses_pre_made_client", func(t *testing.T) {
1719
want := s3.New(s3.Options{Region: "us-east-1"})
18-
api := newAWSs3API(want)
20+
api := newAWSs3API(want, logp.NewNopLogger())
1921
got := api.clientFor("")
2022

2123
if want != got {

0 commit comments

Comments
 (0)