Skip to content

Commit 9a58c4d

Browse files
ZhangxunmtZhangxunmt
committed
create s3 common module
Signed-off-by: Xun Zhang <xunzh@amazon.com>
1 parent 3bf6c95 commit 9a58c4d

55 files changed

Lines changed: 372 additions & 137 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎data-prepper-plugins/s3-common/build.gradle‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
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+
dependencies {
12+
compileOnly 'org.projectlombok:lombok:1.18.30'
13+
annotationProcessor 'org.projectlombok:lombok:1.18.30'
14+
implementation project(path: ':data-prepper-plugins:common')
15+
implementation 'software.amazon.awssdk:sts'
16+
implementation 'software.amazon.awssdk:s3'
17+
implementation 'com.fasterxml.jackson.core:jackson-databind'
18+
implementation 'org.apache.httpcomponents:httpcore:4.4.16'
19+
implementation 'dev.failsafe:failsafe:3.3.2'
20+
implementation 'io.micrometer:micrometer-core'
21+
testImplementation libs.commons.lang3
22+
implementation libs.parquet.common
23+
testImplementation testLibs.bundles.junit
24+
}
25+
26+
jacocoTestCoverageVerification {
27+
dependsOn jacocoTestReport
28+
violationRules {
29+
rule { //in addition to core projects rule
30+
limit {
31+
minimum = 0.90
32+
}
33+
}
34+
}
35+
}

data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/configuration/AwsAuthenticationOptions.java renamed to data-prepper-plugins/s3-common/src/main/java/org/opensearch/dataprepper/plugins/s3/common/config/AwsAuthenticationOptions.java

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,19 @@
11
/*
22
* Copyright OpenSearch Contributors
33
* 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+
*
49
*/
510

6-
package org.opensearch.dataprepper.plugins.source.s3.configuration;
11+
package org.opensearch.dataprepper.plugins.s3.common.config;
712

813
import com.fasterxml.jackson.annotation.JsonProperty;
914
import jakarta.validation.constraints.Size;
10-
import software.amazon.awssdk.arns.Arn;
1115
import software.amazon.awssdk.regions.Region;
16+
import software.amazon.awssdk.arns.Arn;
1217

1318
import java.util.Map;
1419
import java.util.Optional;
@@ -68,4 +73,3 @@ public Map<String, String> getAwsStsHeaderOverrides() {
6873
return awsStsHeaderOverrides;
6974
}
7075
}
71-

‎data-prepper-plugins/s3-common/src/main/java/org/opensearch/dataprepper/plugins/s3/common/config/S3CommonConfig.java‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
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.s3.common.config;
12+
13+
import com.fasterxml.jackson.annotation.JsonProperty;
14+
import jakarta.validation.Valid;
15+
import jakarta.validation.constraints.NotNull;
16+
17+
import java.util.Map;
18+
19+
/**
20+
* Abstract base class for S3-related configurations that require bucket ownership.
21+
*/
22+
public abstract class S3CommonConfig {
23+
@JsonProperty("aws")
24+
@NotNull
25+
@Valid
26+
private AwsAuthenticationOptions awsAuthenticationOptions;
27+
28+
/**
29+
* @return true if bucket ownership validation should be disabled
30+
*/
31+
public abstract boolean isDisableBucketOwnershipValidation();
32+
33+
/**
34+
* @return the default bucket owner account ID, or null if not configured
35+
*/
36+
public abstract String getDefaultBucketOwner();
37+
38+
/**
39+
* @return map of bucket names to owner account IDs, or null/empty if not configured
40+
*/
41+
public abstract Map<String, String> getBucketOwners();
42+
43+
public AwsAuthenticationOptions getAwsAuthenticationOptions() {
44+
return awsAuthenticationOptions;
45+
}
46+
}

data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/ownership/BucketOwnerProvider.java renamed to data-prepper-plugins/s3-common/src/main/java/org/opensearch/dataprepper/plugins/s3/common/ownership/BucketOwnerProvider.java

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,14 @@
11
/*
22
* Copyright OpenSearch Contributors
33
* 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+
*
49
*/
510

6-
package org.opensearch.dataprepper.plugins.source.s3.ownership;
11+
package org.opensearch.dataprepper.plugins.s3.common.ownership;
712

813
import java.util.Optional;
914

data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/ownership/MappedBucketOwnerProvider.java renamed to data-prepper-plugins/s3-common/src/main/java/org/opensearch/dataprepper/plugins/s3/common/ownership/MappedBucketOwnerProvider.java

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,21 +1,30 @@
1-
package org.opensearch.dataprepper.plugins.source.s3.ownership;
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.s3.common.ownership;
212

313
import java.util.HashMap;
414
import java.util.Map;
515
import java.util.Objects;
616
import java.util.Optional;
7-
817
/**
918
* Implements {@link BucketOwnerProvider} using a mapping of bucket
1019
* names to account Ids for the bucket owners. Uses a delegate
1120
* {@link BucketOwnerProvider} as a fallback when the bucket is not
1221
* found in the map.
1322
*/
14-
class MappedBucketOwnerProvider implements BucketOwnerProvider {
23+
public class MappedBucketOwnerProvider implements BucketOwnerProvider {
1524
private final Map<String, String> bucketOwnershipMap;
1625
private final BucketOwnerProvider fallbackProvider;
1726

18-
MappedBucketOwnerProvider(Map<String, String> bucketOwnershipMap, BucketOwnerProvider fallbackProvider) {
27+
public MappedBucketOwnerProvider(Map<String, String> bucketOwnershipMap, BucketOwnerProvider fallbackProvider) {
1928
this.bucketOwnershipMap = new HashMap<>(Objects.requireNonNull(bucketOwnershipMap));
2029
this.fallbackProvider = Objects.requireNonNull(fallbackProvider);
2130
}

data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/ownership/NoOwnershipBucketOwnerProvider.java renamed to data-prepper-plugins/s3-common/src/main/java/org/opensearch/dataprepper/plugins/s3/common/ownership/NoOwnershipBucketOwnerProvider.java

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,22 @@
11
/*
22
* Copyright OpenSearch Contributors
33
* 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+
*
49
*/
510

6-
package org.opensearch.dataprepper.plugins.source.s3.ownership;
11+
package org.opensearch.dataprepper.plugins.s3.common.ownership;
712

813
import java.util.Optional;
914

1015
/**
1116
* An implementation of {@link BucketOwnerProvider} which does not provide
1217
* a bucket owner, effectively skipping owner validation.
1318
*/
14-
class NoOwnershipBucketOwnerProvider implements BucketOwnerProvider {
19+
public class NoOwnershipBucketOwnerProvider implements BucketOwnerProvider{
1520
@Override
1621
public Optional<String> getBucketOwner(final String bucket) {
1722
return Optional.empty();

data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/ownership/StaticBucketOwnerProvider.java renamed to data-prepper-plugins/s3-common/src/main/java/org/opensearch/dataprepper/plugins/s3/common/ownership/StaticBucketOwnerProvider.java

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,19 @@
11
/*
22
* Copyright OpenSearch Contributors
33
* 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+
*
49
*/
510

6-
package org.opensearch.dataprepper.plugins.source.s3.ownership;
11+
package org.opensearch.dataprepper.plugins.s3.common.ownership;
712

813
import java.util.Objects;
914
import java.util.Optional;
1015

11-
/**
12-
* An implementation of {@link BucketOwnerProvider} which provides the
13-
* same owner for all buckets.
14-
*/
15-
class StaticBucketOwnerProvider implements BucketOwnerProvider {
16+
public class StaticBucketOwnerProvider implements BucketOwnerProvider {
1617
private final String accountId;
1718

1819
public StaticBucketOwnerProvider(final String accountId) {

data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3InputFile.java renamed to data-prepper-plugins/s3-common/src/main/java/org/opensearch/dataprepper/plugins/s3/common/source/S3InputFile.java

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,18 @@
1-
package org.opensearch.dataprepper.plugins.source.s3;
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.s3.common.source;
212

313
import org.apache.parquet.io.SeekableInputStream;
414
import org.opensearch.dataprepper.model.io.InputFile;
5-
import org.opensearch.dataprepper.plugins.source.s3.ownership.BucketOwnerProvider;
15+
import org.opensearch.dataprepper.plugins.s3.common.ownership.BucketOwnerProvider;
616
import software.amazon.awssdk.services.s3.S3Client;
717
import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
818
import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
@@ -67,7 +77,7 @@ public Instant getLastModified() {
6777
@Override
6878
public SeekableInputStream newStream() {
6979
return new S3InputStream(
70-
s3Client, s3ObjectReference, bucketOwnerProvider, getMetadata(), s3ObjectPluginMetrics, DEFAULT_RETRY_DELAY, DEFAULT_RETRIES);
80+
s3Client, s3ObjectReference, bucketOwnerProvider, getMetadata(), s3ObjectPluginMetrics, DEFAULT_RETRY_DELAY, DEFAULT_RETRIES);
7181
}
7282

7383
/**

data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3InputStream.java renamed to data-prepper-plugins/s3-common/src/main/java/org/opensearch/dataprepper/plugins/s3/common/source/S3InputStream.java

Lines changed: 28 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,14 @@
1-
package org.opensearch.dataprepper.plugins.source.s3;
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.s3.common.source;
212

313
import com.google.common.base.Preconditions;
414
import com.google.common.io.ByteStreams;
@@ -8,7 +18,7 @@
818
import dev.failsafe.function.CheckedSupplier;
919
import org.apache.http.ConnectionClosedException;
1020
import org.apache.parquet.io.SeekableInputStream;
11-
import org.opensearch.dataprepper.plugins.source.s3.ownership.BucketOwnerProvider;
21+
import org.opensearch.dataprepper.plugins.s3.common.ownership.BucketOwnerProvider;
1222
import org.slf4j.Logger;
1323
import org.slf4j.LoggerFactory;
1424
import software.amazon.awssdk.core.sync.ResponseTransformer;
@@ -29,13 +39,13 @@
2939
import java.util.List;
3040
import java.util.concurrent.atomic.LongAdder;
3141

32-
class S3InputStream extends SeekableInputStream {
42+
public class S3InputStream extends SeekableInputStream {
3343

34-
static final List<Class<? extends Throwable>> RETRYABLE_EXCEPTIONS = List.of(
35-
ConnectionClosedException.class,
36-
EOFException.class,
37-
SocketException.class,
38-
SocketTimeoutException.class
44+
public static final List<Class<? extends Throwable>> RETRYABLE_EXCEPTIONS = List.of(
45+
ConnectionClosedException.class,
46+
EOFException.class,
47+
SocketException.class,
48+
SocketTimeoutException.class
3949
);
4050

4151
private static final int COPY_BUFFER_SIZE = 8192;
@@ -89,23 +99,23 @@ public S3InputStream(
8999
this.bytesCounter = new LongAdder();
90100

91101
this.getObjectRequestBuilder = GetObjectRequest.builder()
92-
.bucket(this.s3ObjectReference.getBucketName())
93-
.key(this.s3ObjectReference.getKey());
102+
.bucket(this.s3ObjectReference.getBucketName())
103+
.key(this.s3ObjectReference.getKey());
94104

95105
bucketOwnerProvider.getBucketOwner(this.s3ObjectReference.getBucketName())
96106
.ifPresent(getObjectRequestBuilder::expectedBucketOwner);
97107

98108
this.retryPolicyReturningByteArray = RetryPolicy.<byte[]>builder()
99-
.handle(RETRYABLE_EXCEPTIONS)
100-
.withDelay(retryDelay)
101-
.withMaxRetries(retries)
102-
.build();
109+
.handle(RETRYABLE_EXCEPTIONS)
110+
.withDelay(retryDelay)
111+
.withMaxRetries(retries)
112+
.build();
103113

104114
this.retryPolicyReturningInteger = RetryPolicy.<Integer>builder()
105-
.handle(RETRYABLE_EXCEPTIONS)
106-
.withDelay(retryDelay)
107-
.withMaxRetries(retries)
108-
.build();
115+
.handle(RETRYABLE_EXCEPTIONS)
116+
.withDelay(retryDelay)
117+
.withMaxRetries(retries)
118+
.build();
109119
}
110120

111121

data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3ObjectPluginMetrics.java renamed to data-prepper-plugins/s3-common/src/main/java/org/opensearch/dataprepper/plugins/s3/common/source/S3ObjectPluginMetrics.java

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,14 @@
11
/*
22
* Copyright OpenSearch Contributors
33
* 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+
*
49
*/
5-
package org.opensearch.dataprepper.plugins.source.s3;
10+
11+
package org.opensearch.dataprepper.plugins.s3.common.source;
612

713
import io.micrometer.core.instrument.Counter;
814
import io.micrometer.core.instrument.DistributionSummary;
@@ -94,4 +100,4 @@ public Counter getS3ObjectsThrottledCounter() {
94100
public Counter getS3ObjectsDeleteFailed() { return s3ObjectsDeleteFailed; }
95101

96102
public Counter getS3ObjectReadFailedCounter() { return s3ObjectReadFailedCounter; }
97-
}
103+
}

0 commit comments

Comments
 (0)