-
Notifications
You must be signed in to change notification settings - Fork 188
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add support for AWS security lake sink as a bucket selector mode in S…
…3 sink (#4846) * dplive1.yaml Signed-off-by: Kondaka <krishkdk@amazon.com> * Delete .github/workflows/static.yml Signed-off-by: Kondaka <krishkdk@amazon.com> * Add support for AWS security lake sink as a bucket selector mode in S3 sink Signed-off-by: Kondaka <krishkdk@amazon.com> * Fixed tests Signed-off-by: Kondaka <krishkdk@amazon.com> * Added javadoc for S3BucketSelector Signed-off-by: Kondaka <krishkdk@amazon.com> * Added new tests for KeyGenerator Signed-off-by: Kondaka <krishkdk@amazon.com> * Added new tests and fixed style errors Signed-off-by: Kondaka <krishkdk@amazon.com> * Addressed review comments Signed-off-by: Kondaka <krishkdk@amazon.com> * Fixed test build failure Signed-off-by: Kondaka <krishkdk@amazon.com> --------- Signed-off-by: Kondaka <krishkdk@amazon.com>
- Loading branch information
Showing
33 changed files
with
488 additions
and
73 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
17 changes: 17 additions & 0 deletions
17
...nk/src/main/java/org/opensearch/dataprepper/plugins/sink/s3/PredefinedObjectMetadata.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,17 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.sink.s3; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonProperty; | ||
| public class PredefinedObjectMetadata { | ||
| @JsonProperty("number_of_objects") | ||
| private String numberOfObjects; | ||
|
|
||
| public String getNumberOfObjects() { | ||
| return numberOfObjects; | ||
| } | ||
|
|
||
| } |
29 changes: 29 additions & 0 deletions
29
...ns/s3-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/s3/S3BucketSelector.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,29 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.sink.s3; | ||
|
|
||
| public interface S3BucketSelector { | ||
| /** | ||
| * initialize - initializes the selector | ||
| * @param s3SinkConfig - s3 sink configuration | ||
| */ | ||
| void initialize(S3SinkConfig s3SinkConfig); | ||
|
|
||
| /** | ||
| * getBucketName - returns the name of the bucket created by the bucket selector | ||
| * | ||
| * @return - bucket name | ||
| */ | ||
| String getBucketName(); | ||
|
|
||
| /** | ||
| * getPathPrefix - returns the prefix to be used for the objects created in the bucket | ||
| * | ||
| * @return path prefix | ||
| */ | ||
| String getPathPrefix(); | ||
| } | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
80 changes: 80 additions & 0 deletions
80
.../src/main/java/org/opensearch/dataprepper/plugins/sink/s3/SecurityLakeBucketSelector.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,80 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.sink.s3; | ||
|
|
||
| import org.apache.commons.lang3.RandomStringUtils; | ||
| import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin; | ||
| import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor; | ||
| import software.amazon.awssdk.services.securitylake.SecurityLakeClient; | ||
| import software.amazon.awssdk.services.securitylake.model.AwsIdentity; | ||
| import software.amazon.awssdk.services.securitylake.model.CreateCustomLogSourceRequest; | ||
| import software.amazon.awssdk.services.securitylake.model.CustomLogSourceProvider; | ||
| import software.amazon.awssdk.services.securitylake.model.CustomLogSourceConfiguration; | ||
| import software.amazon.awssdk.services.securitylake.model.CreateCustomLogSourceResponse; | ||
| import software.amazon.awssdk.services.securitylake.model.CustomLogSourceCrawlerConfiguration; | ||
|
|
||
| import java.time.LocalDate; | ||
| import java.util.List; | ||
|
|
||
| @DataPrepperPlugin(name = "aws_security_lake", pluginType = S3BucketSelector.class, pluginConfigurationType = SecurityLakeBucketSelectorConfig.class) | ||
| public class SecurityLakeBucketSelector implements S3BucketSelector { | ||
| private static final String EXT_PATH = "/ext/"; | ||
| private final SecurityLakeBucketSelectorConfig securityLakeBucketSelectorConfig; | ||
|
|
||
| private S3SinkConfig s3SinkConfig; | ||
|
|
||
| private String pathPrefix; | ||
|
|
||
| private String sourceLocation; | ||
|
|
||
| @DataPrepperPluginConstructor | ||
| public SecurityLakeBucketSelector(final SecurityLakeBucketSelectorConfig securityLakeBucketSelectorConfig) { | ||
| this.securityLakeBucketSelectorConfig = securityLakeBucketSelectorConfig; | ||
| } | ||
|
|
||
| public void initialize(S3SinkConfig s3SinkConfig) { | ||
| this.s3SinkConfig = s3SinkConfig; | ||
| SecurityLakeClient securityLakeClient = SecurityLakeClient.create(); | ||
| String arn = s3SinkConfig.getAwsAuthenticationOptions().getAwsStsRoleArn(); | ||
| String principal = arn.split(":")[4]; | ||
| String sourceName = securityLakeBucketSelectorConfig.getSourceName() != null ? securityLakeBucketSelectorConfig.getSourceName() : RandomStringUtils.randomAlphabetic(7); | ||
| CreateCustomLogSourceResponse response = | ||
| securityLakeClient.createCustomLogSource( | ||
| CreateCustomLogSourceRequest.builder() | ||
| .sourceName(sourceName+RandomStringUtils.randomAlphabetic(4)) | ||
| .eventClasses(List.of(securityLakeBucketSelectorConfig.getLogClass())) | ||
| .sourceVersion(securityLakeBucketSelectorConfig.getSourceVersion()) | ||
| .configuration(CustomLogSourceConfiguration.builder() | ||
| .crawlerConfiguration(CustomLogSourceCrawlerConfiguration.builder() | ||
| .roleArn(arn) | ||
| .build()) | ||
| .providerIdentity(AwsIdentity.builder() | ||
| .externalId(securityLakeBucketSelectorConfig.getExternalId()) | ||
| .principal(principal) | ||
| .build()) | ||
| .build()) | ||
| .build()); | ||
| CustomLogSourceProvider provider = response.source().provider(); | ||
| this.sourceLocation = provider.location(); | ||
| final String region=s3SinkConfig.getAwsAuthenticationOptions().getAwsRegion().toString(); | ||
| final String accountId=arn.split(":")[4]; | ||
|
|
||
| final LocalDate now = LocalDate.now(); | ||
| final String eventDay = String.format("%d%02d%02d", now.getYear(), now.getMonthValue(), now.getDayOfMonth()); | ||
| int locIndex = sourceLocation.indexOf(EXT_PATH); | ||
| pathPrefix = String.format("%sregion=%s/accountId=%s/eventDay=%s/",sourceLocation.substring(locIndex+1), region, accountId, eventDay); | ||
| } | ||
|
|
||
| public String getPathPrefix() { | ||
| return pathPrefix; | ||
| } | ||
|
|
||
| @Override | ||
| public String getBucketName() { | ||
| int locIndex = sourceLocation.indexOf(EXT_PATH); | ||
| return sourceLocation.substring(EXT_PATH.length(), locIndex); | ||
| } | ||
| } |
44 changes: 44 additions & 0 deletions
44
...ain/java/org/opensearch/dataprepper/plugins/sink/s3/SecurityLakeBucketSelectorConfig.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.sink.s3; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonProperty; | ||
|
|
||
| public class SecurityLakeBucketSelectorConfig { | ||
| static final String DEFAULT_SOURCE_VERSION = "1.0"; | ||
|
|
||
| static final String DEFAULT_EXTERNAL_ID = "extid"; | ||
|
|
||
| @JsonProperty("source_name") | ||
| private String sourceName; | ||
|
|
||
| @JsonProperty("source_version") | ||
| private String sourceVersion = DEFAULT_SOURCE_VERSION; | ||
|
|
||
| @JsonProperty("external_id") | ||
| private String externalId = DEFAULT_EXTERNAL_ID; | ||
|
|
||
| @JsonProperty("log_class") | ||
| private String logClass; | ||
|
|
||
|
|
||
| public String getSourceName() { | ||
| return sourceName; | ||
| } | ||
|
|
||
| public String getSourceVersion() { | ||
| return sourceVersion; | ||
| } | ||
|
|
||
| public String getExternalId() { | ||
| return externalId; | ||
| } | ||
|
|
||
| public String getLogClass() { | ||
| return logClass; | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.