Skip to content

Commit 59d45a6

Browse files
lawofcycleslawofcycles
authored
Add top-level filters option for S3 source (#6735)
Add top-level filters option for S3 source Add a top-level filters configuration to the S3 source that applies include_prefix and exclude_suffix filtering for both SQS and scan modes. Previously, key path filters were only available under scan bucket options, making it impossible to filter S3 objects when using SQS notifications. The new filters option uses the same bucket name keyed Map pattern as bucket_owners. Top-level filters and scan bucket-level filters cannot be used together, as the top-level filters are intended to eventually replace the scan bucket-level filters. Signed-off-by: Sotaro Hikita <bering1814@gmail.com>
1 parent 4990f26 commit 59d45a6

8 files changed

Lines changed: 261 additions & 5 deletions

File tree

‎data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3ObjectKeyFilter.java‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*
5+
* The OpenSearch Contributors require contributions made to
6+
* this file be licensed under the Apache-2.0 license or a
7+
* compatible open source license.
8+
*
9+
*/
10+
11+
package org.opensearch.dataprepper.plugins.source.s3;
12+
13+
import org.opensearch.dataprepper.plugins.source.s3.configuration.S3ScanKeyPathOption;
14+
15+
import java.util.List;
16+
import java.util.Map;
17+
18+
/**
19+
* Filters S3 objects by key using top-level include_prefix and exclude_suffix options.
20+
*/
21+
public class S3ObjectKeyFilter {
22+
23+
private final Map<String, S3ScanKeyPathOption> filters;
24+
25+
public S3ObjectKeyFilter(final Map<String, S3ScanKeyPathOption> filters) {
26+
this.filters = filters;
27+
}
28+
29+
/**
30+
* Returns true if the object key matches the filters for the given bucket,
31+
* or if no filters are configured for the bucket.
32+
*/
33+
public boolean isKeyMatchingFilters(final String bucketName, final String objectKey) {
34+
if (filters == null || filters.isEmpty()) {
35+
return true;
36+
}
37+
38+
final S3ScanKeyPathOption keyPathOption = filters.get(bucketName);
39+
if (keyPathOption == null) {
40+
return true;
41+
}
42+
43+
final List<String> includePrefixes = keyPathOption.getS3scanIncludePrefixOptions();
44+
if (includePrefixes != null && !includePrefixes.isEmpty()
45+
&& includePrefixes.stream().noneMatch(objectKey::startsWith)) {
46+
return false;
47+
}
48+
49+
final List<String> excludeSuffixes = keyPathOption.getS3ScanExcludeSuffixOptions();
50+
if (excludeSuffixes != null && !excludeSuffixes.isEmpty()
51+
&& excludeSuffixes.stream().anyMatch(objectKey::endsWith)) {
52+
return false;
53+
}
54+
55+
return true;
56+
}
57+
}

‎data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3ScanPartitionCreationSupplier.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,13 +56,16 @@ public class S3ScanPartitionCreationSupplier implements Function<Map<String, Obj
5656

5757
private final SourceCoordinator<S3SourceProgressState> sourceCoordinator;
5858

59+
private final S3ObjectKeyFilter objectFilteringHelper;
60+
5961
public S3ScanPartitionCreationSupplier(final S3Client s3Client,
6062
final BucketOwnerProvider bucketOwnerProvider,
6163
final List<ScanOptions> scanOptionsList,
6264
final S3ScanSchedulingOptions schedulingOptions,
6365
final FolderPartitioningOptions folderPartitioningOptions,
6466
final boolean deleteS3ObjectsOnRead,
65-
final SourceCoordinator<S3SourceProgressState> sourceCoordinator) {
67+
final SourceCoordinator<S3SourceProgressState> sourceCoordinator,
68+
final S3ObjectKeyFilter objectFilteringHelper) {
6669

6770
this.s3Client = s3Client;
6871
this.bucketOwnerProvider = bucketOwnerProvider;
@@ -71,6 +74,7 @@ public S3ScanPartitionCreationSupplier(final S3Client s3Client,
7174
this.folderPartitioningOptions = folderPartitioningOptions;
7275
this.deleteS3ObjectsOnRead = deleteS3ObjectsOnRead;
7376
this.sourceCoordinator = sourceCoordinator;
77+
this.objectFilteringHelper = objectFilteringHelper;
7478
}
7579

7680
@Override
@@ -145,6 +149,7 @@ private void createFilteredS3ObjectPartitionsForBucket(final List<String> exclud
145149
.filter(keyTimestampPair -> !keyTimestampPair.left().endsWith("/"))
146150
.filter(keyTimestampPair -> excludeKeyPaths.stream()
147151
.noneMatch(excludeItem -> keyTimestampPair.left().endsWith(excludeItem)))
152+
.filter(keyTimestampPair -> objectFilteringHelper.isKeyMatchingFilters(bucket, keyTimestampPair.left()))
148153
.filter(keyTimestampPair -> isKeyMatchedBetweenTimeRange(keyTimestampPair.right(), startDateTime, endDateTime, isFirstScan))
149154
.map(Pair::left)
150155
.map(objectKey -> PartitionIdentifier.builder().withPartitionKey(String.format(BUCKET_OBJECT_PARTITION_KEY_FORMAT, bucket, objectKey)).build())

‎data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3SourceConfig.java‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,10 @@
2222
import org.opensearch.dataprepper.plugins.source.s3.configuration.S3SelectOptions;
2323
import org.opensearch.dataprepper.plugins.source.s3.configuration.SqsOptions;
2424
import org.opensearch.dataprepper.plugins.source.s3.configuration.S3DataSelection;
25+
import org.opensearch.dataprepper.plugins.source.s3.configuration.S3ScanKeyPathOption;
2526

2627
import java.time.Duration;
28+
import java.util.Collections;
2729
import java.util.Map;
2830

2931
public class S3SourceConfig {
@@ -104,6 +106,10 @@ public class S3SourceConfig {
104106
@JsonProperty("data_selection")
105107
private S3DataSelection dataSelection = S3DataSelection.DATA_AND_METADATA;
106108

109+
@JsonProperty("filters")
110+
@Valid
111+
private Map<String, S3ScanKeyPathOption> filters;
112+
107113
@AssertTrue(message = "A codec is required for reading objects.")
108114
boolean isCodecProvidedWhenNeeded() {
109115
if(s3SelectOptions == null)
@@ -139,6 +145,19 @@ boolean isS3SelectNotUsingDeleteS3ObjectsOnRead() {
139145
return true;
140146
}
141147

148+
@AssertTrue(message = "Top-level filters cannot be used together with scan bucket-level filter. Use one or the other.")
149+
boolean isFiltersNotUsedWithScanBucketFilter() {
150+
if (filters == null || filters.isEmpty()) {
151+
return true;
152+
}
153+
if (s3ScanScanOptions == null || s3ScanScanOptions.getBuckets() == null) {
154+
return true;
155+
}
156+
return s3ScanScanOptions.getBuckets().stream()
157+
.map(bucket -> bucket.getS3ScanBucketOption())
158+
.noneMatch(option -> option != null && option.getS3ScanFilter() != null);
159+
}
160+
142161
public NotificationTypeOption getNotificationType() {
143162
return notificationType;
144163
}
@@ -222,4 +241,8 @@ public boolean isDeleteS3MetadataInEvent() {
222241
public S3DataSelection getDataSelection() {
223242
return dataSelection;
224243
}
244+
245+
public Map<String, S3ScanKeyPathOption> getFilters() {
246+
return filters != null ? filters : Collections.emptyMap();
247+
}
225248
}

‎data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/ScanObjectWorker.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -168,7 +168,7 @@ public ScanObjectWorker(final S3Client s3Client,
168168
this.folderPartitioningOptions = s3SourceConfig.getS3ScanScanOptions().getPartitioningOptions();
169169
this.acknowledgmentSetTimeout = s3SourceConfig.getS3ScanScanOptions().getAcknowledgmentTimeout();
170170

171-
this.partitionCreationSupplier = new S3ScanPartitionCreationSupplier(s3Client, bucketOwnerProvider, scanOptionsBuilderList, s3ScanSchedulingOptions, s3SourceConfig.getS3ScanScanOptions().getPartitioningOptions(), s3SourceConfig.isDeleteS3ObjectsOnRead(), sourceCoordinator);
171+
this.partitionCreationSupplier = new S3ScanPartitionCreationSupplier(s3Client, bucketOwnerProvider, scanOptionsBuilderList, s3ScanSchedulingOptions, s3SourceConfig.getS3ScanScanOptions().getPartitioningOptions(), s3SourceConfig.isDeleteS3ObjectsOnRead(), sourceCoordinator, new S3ObjectKeyFilter(s3SourceConfig.getFilters()));
172172
this.acknowledgmentsRemainingForPartitions = new ConcurrentHashMap<>();
173173
this.objectsToDeleteForAcknowledgmentSets = new ConcurrentHashMap<>();
174174
}

‎data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/SqsWorker.java‎

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -67,13 +67,15 @@ public class SqsWorker implements Runnable {
6767
static final String SQS_MESSAGE_ACCESS_DENIED_METRIC_NAME = "sqsMessagesAccessDenied";
6868
static final String SQS_MESSAGE_THROTTLED_METRIC_NAME = "sqsMessagesThrottled";
6969
static final String SQS_RESOURCE_NOT_FOUND_METRIC_NAME = "sqsResourceNotFound";
70+
static final String S3_OBJECTS_FILTERED_METRIC_NAME = "s3ObjectsFiltered";
7071

7172
private final S3SourceConfig s3SourceConfig;
7273
private final SqsClient sqsClient;
7374
private final S3Service s3Service;
7475
private final SqsOptions sqsOptions;
7576
private final S3EventFilter objectCreatedFilter;
7677
private final S3EventFilter evenBridgeObjectCreatedFilter;
78+
private final S3ObjectKeyFilter objectFilteringHelper;
7779
private final Counter sqsMessagesReceivedCounter;
7880
private final Counter sqsReceiveMessagesFailedCounter;
7981
private final Counter sqsMessagesDeletedCounter;
@@ -86,6 +88,7 @@ public class SqsWorker implements Runnable {
8688
private final Counter sqsMessageAccessDeniedCounter;
8789
private final Counter sqsMessageThrottledCounter;
8890
private final Counter sqsResourceNotFoundCounter;
91+
private final Counter s3ObjectsFilteredCounter;
8992
private final Timer sqsMessageDelayTimer;
9093
private final Backoff standardBackoff;
9194
private final SqsMessageParser sqsMessageParser;
@@ -110,6 +113,7 @@ public SqsWorker(final AcknowledgementSetManager acknowledgementSetManager,
110113
sqsOptions = s3SourceConfig.getSqsOptions();
111114
objectCreatedFilter = new S3ObjectCreatedFilter();
112115
evenBridgeObjectCreatedFilter = new EventBridgeObjectCreatedFilter();
116+
objectFilteringHelper = new S3ObjectKeyFilter(s3SourceConfig.getFilters());
113117
sqsMessageParser = new SqsMessageParser(s3SourceConfig);
114118
failedAttemptCount = 0;
115119
parsedMessageVisibilityTimesMap = new HashMap<>();
@@ -126,6 +130,7 @@ public SqsWorker(final AcknowledgementSetManager acknowledgementSetManager,
126130
sqsMessageAccessDeniedCounter = pluginMetrics.counter(SQS_MESSAGE_ACCESS_DENIED_METRIC_NAME);
127131
sqsMessageThrottledCounter = pluginMetrics.counter(SQS_MESSAGE_THROTTLED_METRIC_NAME);
128132
sqsResourceNotFoundCounter = pluginMetrics.counter(SQS_RESOURCE_NOT_FOUND_METRIC_NAME);
133+
s3ObjectsFilteredCounter = pluginMetrics.counter(S3_OBJECTS_FILTERED_METRIC_NAME);
129134
}
130135

131136
@Override
@@ -234,11 +239,11 @@ private List<DeleteMessageBatchRequestEntry> processS3EventNotificationRecords(f
234239
if (s3SourceConfig.getNotificationSource().equals(NotificationSourceOption.S3)
235240
&& !parsedMessage.isEmptyNotification()
236241
&& isS3EventNameCreated(parsedMessage)) {
237-
parsedMessagesToRead.add(parsedMessage);
242+
addParsedMessageByFilter(parsedMessage, parsedMessagesToRead, deleteMessageBatchRequestEntryCollection);
238243
}
239244
else if (s3SourceConfig.getNotificationSource().equals(NotificationSourceOption.EVENTBRIDGE)
240245
&& isEventBridgeEventTypeCreated(parsedMessage)) {
241-
parsedMessagesToRead.add(parsedMessage);
246+
addParsedMessageByFilter(parsedMessage, parsedMessagesToRead, deleteMessageBatchRequestEntryCollection);
242247
}
243248
else {
244249
// TODO: Delete these only if on_error is configured to delete_messages.
@@ -454,6 +459,19 @@ private boolean isEventBridgeEventTypeCreated(final ParsedMessage parsedMessage)
454459
return evenBridgeObjectCreatedFilter.filter(parsedMessage).isPresent();
455460
}
456461

462+
private void addParsedMessageByFilter(final ParsedMessage parsedMessage,
463+
final List<ParsedMessage> parsedMessagesToRead,
464+
final List<DeleteMessageBatchRequestEntry> deleteMessageBatchRequestEntries) {
465+
if (objectFilteringHelper.isKeyMatchingFilters(parsedMessage.getBucketName(), parsedMessage.getObjectKey())) {
466+
parsedMessagesToRead.add(parsedMessage);
467+
} else {
468+
LOG.debug("S3 object {} in bucket {} did not match configured filters. Deleting SQS message.",
469+
parsedMessage.getObjectKey(), parsedMessage.getBucketName());
470+
s3ObjectsFilteredCounter.increment();
471+
deleteMessageBatchRequestEntries.add(buildDeleteMessageBatchRequestEntry(parsedMessage.getMessage()));
472+
}
473+
}
474+
457475
private S3ObjectReference populateS3Reference(final String bucketName, final String objectKey) {
458476
return S3ObjectReference
459477
.bucketAndKey(bucketName, objectKey)

‎data-prepper-plugins/s3-source/src/test/java/org/opensearch/dataprepper/plugins/source/s3/S3ObjectKeyFilterTest.java‎

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*
5+
* The OpenSearch Contributors require contributions made to
6+
* this file be licensed under the Apache-2.0 license or a
7+
* compatible open source license.
8+
*
9+
*/
10+
11+
package org.opensearch.dataprepper.plugins.source.s3;
12+
13+
import org.junit.jupiter.api.Test;
14+
import org.junit.jupiter.params.ParameterizedTest;
15+
import org.junit.jupiter.params.provider.CsvSource;
16+
import org.opensearch.dataprepper.plugins.source.s3.configuration.S3ScanKeyPathOption;
17+
18+
import java.util.Collections;
19+
import java.util.List;
20+
import java.util.Map;
21+
22+
import static org.hamcrest.CoreMatchers.equalTo;
23+
import static org.hamcrest.MatcherAssert.assertThat;
24+
import static org.mockito.Mockito.mock;
25+
import static org.mockito.Mockito.when;
26+
27+
class S3ObjectKeyFilterTest {
28+
29+
@Test
30+
void isKeyMatchingFilters_returns_true_when_filters_is_empty() {
31+
final S3ObjectKeyFilter filter = new S3ObjectKeyFilter(Collections.emptyMap());
32+
assertThat(filter.isKeyMatchingFilters("my-bucket", "any/key.json"), equalTo(true));
33+
}
34+
35+
@Test
36+
void isKeyMatchingFilters_returns_true_when_filters_is_null() {
37+
final S3ObjectKeyFilter filter = new S3ObjectKeyFilter(null);
38+
assertThat(filter.isKeyMatchingFilters("my-bucket", "any/key.json"), equalTo(true));
39+
}
40+
41+
@Test
42+
void isKeyMatchingFilters_returns_true_when_bucket_not_in_filters() {
43+
final S3ScanKeyPathOption keyPathOption = mock(S3ScanKeyPathOption.class);
44+
when(keyPathOption.getS3scanIncludePrefixOptions()).thenReturn(List.of("assets/"));
45+
final S3ObjectKeyFilter filter = new S3ObjectKeyFilter(Map.of("other-bucket", keyPathOption));
46+
47+
assertThat(filter.isKeyMatchingFilters("my-bucket", "any/key.json"), equalTo(true));
48+
}
49+
50+
@ParameterizedTest
51+
@CsvSource({
52+
"assets/image.png, true",
53+
"logs/app.log, false"
54+
})
55+
void isKeyMatchingFilters_with_include_prefix(final String objectKey, final boolean expected) {
56+
final S3ScanKeyPathOption keyPathOption = mock(S3ScanKeyPathOption.class);
57+
when(keyPathOption.getS3scanIncludePrefixOptions()).thenReturn(List.of("assets/"));
58+
when(keyPathOption.getS3ScanExcludeSuffixOptions()).thenReturn(null);
59+
final S3ObjectKeyFilter filter = new S3ObjectKeyFilter(Map.of("my-bucket", keyPathOption));
60+
61+
assertThat(filter.isKeyMatchingFilters("my-bucket", objectKey), equalTo(expected));
62+
}
63+
64+
@ParameterizedTest
65+
@CsvSource({
66+
"assets/photo.jpg, false",
67+
"assets/data.json, true"
68+
})
69+
void isKeyMatchingFilters_with_exclude_suffix(final String objectKey, final boolean expected) {
70+
final S3ScanKeyPathOption keyPathOption = mock(S3ScanKeyPathOption.class);
71+
when(keyPathOption.getS3scanIncludePrefixOptions()).thenReturn(null);
72+
when(keyPathOption.getS3ScanExcludeSuffixOptions()).thenReturn(List.of(".jpg", ".xml"));
73+
final S3ObjectKeyFilter filter = new S3ObjectKeyFilter(Map.of("my-bucket", keyPathOption));
74+
75+
assertThat(filter.isKeyMatchingFilters("my-bucket", objectKey), equalTo(expected));
76+
}
77+
78+
@ParameterizedTest
79+
@CsvSource({
80+
"assets/data.json, true",
81+
"assets/photo.jpg, false",
82+
"logs/app.log, false"
83+
})
84+
void isKeyMatchingFilters_with_both_include_prefix_and_exclude_suffix(final String objectKey, final boolean expected) {
85+
final S3ScanKeyPathOption keyPathOption = mock(S3ScanKeyPathOption.class);
86+
when(keyPathOption.getS3scanIncludePrefixOptions()).thenReturn(List.of("assets/"));
87+
when(keyPathOption.getS3ScanExcludeSuffixOptions()).thenReturn(List.of(".jpg"));
88+
final S3ObjectKeyFilter filter = new S3ObjectKeyFilter(Map.of("my-bucket", keyPathOption));
89+
90+
assertThat(filter.isKeyMatchingFilters("my-bucket", objectKey), equalTo(expected));
91+
}
92+
93+
@Test
94+
void isKeyMatchingFilters_returns_true_when_key_matches_any_include_prefix() {
95+
final S3ScanKeyPathOption keyPathOption = mock(S3ScanKeyPathOption.class);
96+
when(keyPathOption.getS3scanIncludePrefixOptions()).thenReturn(List.of("assets/", "data/"));
97+
final S3ObjectKeyFilter filter = new S3ObjectKeyFilter(Map.of("my-bucket", keyPathOption));
98+
99+
assertThat(filter.isKeyMatchingFilters("my-bucket", "data/file.csv"), equalTo(true));
100+
}
101+
}

‎data-prepper-plugins/s3-source/src/test/java/org/opensearch/dataprepper/plugins/source/s3/S3ScanPartitionCreationSupplierTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -86,7 +86,7 @@ void setup() {
8686

8787

8888
private Function<Map<String, Object>, List<PartitionIdentifier>> createObjectUnderTest() {
89-
return new S3ScanPartitionCreationSupplier(s3Client, bucketOwnerProvider, scanOptionsList, schedulingOptions, folderPartitioningOptions, isDeleteS3ObjectsOnRead, sourceCoordinator);
89+
return new S3ScanPartitionCreationSupplier(s3Client, bucketOwnerProvider, scanOptionsList, schedulingOptions, folderPartitioningOptions, isDeleteS3ObjectsOnRead, sourceCoordinator, new S3ObjectKeyFilter(Collections.emptyMap()));
9090
}
9191

9292
@Test

0 commit comments

Comments
 (0)