add ml merge processor

Signed-off-by: Xun Zhang <xunzh@amazon.com>

Author: Zhangxunmt

data-prepper-plugins/ml-merge-processor/README.md

@@ -0,0 +1,34 @@
+
+# ML Merge Processor
+
+This plugin enables you to merge data from a S3 file with source data from your Data Prepper pipeline.
+
+## Usage
+```aidl
+ml_merge-pipeline:
+...
+  processor:
+    - ml_merge:
+          fields:
+            - to_key: field_B
+              from_key: /modelInput/inputText
+          merge_all_fields: true              # Whether to merge all fields from input source
+          retry_on_failure: 3                 # Retry attempts for failed requests
+          tags_on_failure: ["lookup_failed"]  # Tags for failed events
+          # S3 source configuration
+          source_path_prefix: "s3://bucket/input/"  # Base path for source/input files
+          result_file_suffix: ".out"               # Suffix to identify result files
+          correlation_field: "recordId"            # Field used to match related records
+```
+`fields` as the fields to be merged into the pipeline.
+`merge_all_fields` as the flag to merge all data into the pipeline.
+`source_path_prefix` as the Base path for source/input files
+`correlation_field` as the field name used to match related records
+
+# Metrics
+
+### Counter
+
+## Developer Guide
+
+The integration tests for this plugin do not run as part of the Data Prepper build.

data-prepper-plugins/ml-merge-processor/build.gradle

@@ -0,0 +1,54 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+dependencies {
+    implementation project(path: ':data-prepper-plugins:common')
+    implementation project(':data-prepper-plugins:aws-plugin-api')
+    implementation 'software.amazon.awssdk:sdk-core'
+    implementation 'software.amazon.awssdk:sts'
+    implementation 'io.micrometer:micrometer-core'
+    implementation 'org.json:json'
+    implementation 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310'
+    implementation 'org.projectlombok:lombok:1.18.22'
+    annotationProcessor 'org.projectlombok:lombok:1.18.20'
+    implementation 'software.amazon.awssdk:s3'
+    testImplementation project(':data-prepper-test:test-event')
+    testImplementation testLibs.slf4j.simple
+}
+
+test {
+    useJUnitPlatform()
+}
+
+sourceSets {
+    integrationTest {
+        java {
+            compileClasspath += main.output + test.output
+            runtimeClasspath += main.output + test.output
+            srcDir file('src/integrationTest/java')
+        }
+        resources.srcDir file('src/integrationTest/resources')
+    }
+}
+
+configurations {
+    integrationTestImplementation.extendsFrom testImplementation
+    integrationTestRuntime.extendsFrom testRuntime
+}
+
+task integrationTest(type: Test) {
+    group = 'verification'
+    testClassesDirs = sourceSets.integrationTest.output.classesDirs
+
+    useJUnitPlatform()
+
+    classpath = sourceSets.integrationTest.runtimeClasspath
+
+    systemProperty 'log4j.configurationFile', 'src/test/resources/log4j2.properties'
+
+    filter {
+        includeTestsMatching '*IT'
+    }
+}

data-prepper-plugins/ml-merge-processor/src/main/java/org/opensearch/dataprepper/plugins/ml_merge/common/client/S3ClientFactory.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.ml_merge.common.client;
+
+import org.opensearch.dataprepper.aws.api.AwsCredentialsOptions;
+import org.opensearch.dataprepper.aws.api.AwsCredentialsSupplier;
+import org.opensearch.dataprepper.plugins.ml_merge.common.config.AwsAuthenticationOptions;
+import org.opensearch.dataprepper.plugins.ml_merge.processor.MLMergeProcessorConfig;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+
+public class S3ClientFactory {
+
+    public static S3Client createS3Client(final MLMergeProcessorConfig mlProcessorConfig, final AwsCredentialsSupplier awsCredentialsSupplier) {
+        final AwsCredentialsOptions awsCredentialsOptions = convertToCredentialsOptions(
+                mlProcessorConfig.getAwsAuthenticationOptions());
+        final Region region = mlProcessorConfig.getAwsAuthenticationOptions().getAwsRegion();
+        final AwsCredentialsProvider awsCredentialsProvider = awsCredentialsSupplier.getProvider(
+                awsCredentialsOptions);
+
+        return S3Client.builder()
+                .region(region)
+                .credentialsProvider(awsCredentialsProvider)
+                .overrideConfiguration(ClientOverrideConfiguration.builder()
+                        .retryPolicy(retryPolicy -> retryPolicy.numRetries(5).build())
+                        .build())
+                .build();
+    }
+
+    public static AwsCredentialsOptions convertToCredentialsOptions(
+            final AwsAuthenticationOptions awsAuthenticationOptions) {
+        if (awsAuthenticationOptions == null || awsAuthenticationOptions.getAwsStsRoleArn() == null) {
+            return AwsCredentialsOptions.defaultOptionsWithDefaultCredentialsProvider();
+        }
+        return AwsCredentialsOptions.builder()
+                .withRegion(awsAuthenticationOptions.getAwsRegion())
+                .withStsRoleArn(awsAuthenticationOptions.getAwsStsRoleArn())
+                .withStsExternalId(awsAuthenticationOptions.getAwsStsExternalId())
+                .withStsHeaderOverrides(awsAuthenticationOptions.getAwsStsHeaderOverrides())
+                .build();
+    }
+}

data-prepper-plugins/ml-merge-processor/src/main/java/org/opensearch/dataprepper/plugins/ml_merge/common/config/AwsAuthenticationOptions.java

@@ -0,0 +1,42 @@
+package org.opensearch.dataprepper.plugins.ml_merge.common.config;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import jakarta.validation.constraints.Size;
+import software.amazon.awssdk.regions.Region;
+
+import java.util.Map;
+
+public class AwsAuthenticationOptions {
+    @JsonProperty("region")
+    @Size(min = 1, message = "Region cannot be empty string")
+    private String awsRegion;
+
+    @JsonProperty("sts_role_arn")
+    @Size(min = 20, max = 2048, message = "awsStsRoleArn length should be between 1 and 2048 characters")
+    private String awsStsRoleArn;
+
+    @JsonProperty("sts_external_id")
+    @Size(min = 2, max = 1224, message = "awsStsExternalId length should be between 2 and 1224 characters")
+    private String awsStsExternalId;
+
+    @JsonProperty("sts_header_overrides")
+    @Size(max = 5, message = "sts_header_overrides supports a maximum of 5 headers to override")
+    private Map<String, String> awsStsHeaderOverrides;
+
+    public Region getAwsRegion() {
+        return awsRegion != null ? Region.of(awsRegion) : null;
+    }
+
+    public String getAwsStsRoleArn() {
+        return awsStsRoleArn;
+    }
+
+    public String getAwsStsExternalId() {
+        return awsStsExternalId;
+    }
+
+    public Map<String, String> getAwsStsHeaderOverrides() {
+        return awsStsHeaderOverrides;
+    }
+
+}