Add configurable path_prefix parameter (#6674)

Introduce an optional `path_prefix` sink parameter to support OpenSearch
instances served behind a reverse proxy under a custom subdirectory.

Resolves #6654

Signed-off-by: Luis Pigueiras <luis.pigueiras@cern.ch>

Author: Pigueiras

data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/ConnectionConfiguration.java

@@ -115,6 +115,7 @@ public class ConnectionConfiguration {
   private final String awsStsExternalId;
   private final Map<String, String> awsStsHeaderOverrides;
   private final Optional<String> proxy;
+  private final String pathPrefix;
   private final boolean serverless;
   private final String serverlessNetworkPolicyName;
   private final String serverlessCollectionName;
@@ -160,6 +161,10 @@ Optional<String> getProxy() {
     return proxy;
   }
 
+  String getPathPrefix() {
+    return pathPrefix;
+  }
+
   Integer getSocketTimeout() {
     return socketTimeout;
   }
@@ -219,6 +224,7 @@ private ConnectionConfiguration(final Builder builder) {
     this.awsStsExternalId = builder.awsStsExternalId;
     this.awsStsHeaderOverrides = builder.awsStsHeaderOverrides;
     this.proxy = builder.proxy;
+    this.pathPrefix = builder.pathPrefix;
     this.serverless = builder.serverless;
     this.serverlessNetworkPolicyName = builder.serverlessNetworkPolicyName;
     this.serverlessCollectionName = builder.serverlessCollectionName;
@@ -312,6 +318,11 @@ public static ConnectionConfiguration readConnectionConfiguration(final OpenSear
       builder = builder.withProxy(proxy);
     }
 
+    final String pathPrefix = openSearchSinkConfig.getPathPrefix();
+    if (pathPrefix != null) {
+      builder = builder.withPathPrefix(pathPrefix);
+    }
+
     final boolean requestCompressionEnabled = openSearchSinkConfig.getEnableRequestCompression();
     builder = builder.withRequestCompressionEnabled(requestCompressionEnabled);
 
@@ -327,6 +338,9 @@ public RestHighLevelClient createClient(AwsCredentialsSupplier awsCredentialsSup
       i++;
     }
     final RestClientBuilder restClientBuilder = RestClient.builder(httpHosts);
+    if (pathPrefix != null) {
+      restClientBuilder.setPathPrefix(pathPrefix);
+    }
     /*
      * Given that this is a patch release, we will support only the IAM based access policy AES domains.
      * We will not support FGAC and Custom endpoint domains. This will be followed in the next version.
@@ -654,6 +668,7 @@ public static class Builder {
     private String awsStsExternalId;
     private Map<String, String> awsStsHeaderOverrides;
     private Optional<String> proxy = Optional.empty();
+    private String pathPrefix;
     private String pipelineName;
     private boolean serverless;
     private String serverlessNetworkPolicyName;
@@ -791,6 +806,11 @@ public Builder withProxy(final String proxy) {
       return this;
     }
 
+    public Builder withPathPrefix(final String pathPrefix) {
+      this.pathPrefix = pathPrefix;
+      return this;
+    }
+
     public Builder withServerless(boolean serverless) {
       this.serverless = serverless;
       return this;

data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/configuration/OpenSearchSinkConfig.java

@@ -81,6 +81,10 @@ public boolean isAuthConfigValid() {
     @JsonProperty("proxy")
     private String proxy = null;
 
+    @Getter
+    @JsonProperty("path_prefix")
+    private String pathPrefix = null;
+
     @Getter
     @JsonProperty("distribution_version")
     private String distributionVersion = DistributionVersion.DEFAULT.getVersion();

data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/ConnectionConfigurationTests.java

@@ -108,6 +108,7 @@ void testReadConnectionConfigurationDefault() throws JsonProcessingException {
         assertNull(connectionConfiguration.getCertPath());
         assertNull(connectionConfiguration.getConnectTimeout());
         assertNull(connectionConfiguration.getSocketTimeout());
+        assertNull(connectionConfiguration.getPathPrefix());
         assertTrue(connectionConfiguration.isRequestCompressionEnabled());
     }
 
@@ -143,6 +144,17 @@ void testCreateClientDefault() throws IOException {
         client.close();
     }
 
+    @Test
+    void testReadConnectionConfigurationWithPathPrefix() throws JsonProcessingException {
+        final Map<String, Object> configMetadata = generateConfigurationMetadata(
+                TEST_HOSTS, null, null, null, null, false, null, null, null, false);
+        configMetadata.put("path_prefix", "/os");
+        final OpenSearchSinkConfig openSearchSinkConfig = getOpenSearchSinkConfigByConfigMetadata(configMetadata);
+        final ConnectionConfiguration connectionConfiguration =
+                ConnectionConfiguration.readConnectionConfiguration(openSearchSinkConfig);
+        assertEquals("/os", connectionConfiguration.getPathPrefix());
+    }
+
     @Test
     void testCreateOpenSearchClientDefault() throws IOException {
         final OpenSearchSinkConfig openSearchSinkConfig = generateOpenSearchSinkConfig(

data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSinkConfigurationTests.java

@@ -31,6 +31,7 @@ public class OpenSearchSinkConfigurationTests {
     private static final String INVALID_ACTIONS_WITH_EXPRESSION_CONFIG = "invalid-actions-with-expression";
     private static final String CREATE_ACTION_CONFIG = "create-action";
     private static final String CREATE_ACTIONS_WITH_EXPRESSION_CONFIG = "create-actions-with-expression";
+    private static final String VALID_SINK_WITH_PATH_PREFIX_CONFIG = "valid-sink-with-path-prefix";
     private ExpressionEvaluator expressionEvaluator;
 
     ObjectMapper objectMapper;
@@ -94,6 +95,13 @@ public void testReadESConfigWithBulkActionCreateExpression() throws IOException
         assertNotNull(openSearchSinkConfiguration.getRetryConfiguration());
     }
 
+    @Test
+    public void testReadOSConfigWithPathPrefix() throws IOException {
+        final OpenSearchSinkConfig openSearchSinkConfig = generateOpenSearchSinkConfig(VALID_SINK_WITH_PATH_PREFIX_CONFIG);
+
+        assertEquals("/os", openSearchSinkConfig.getPathPrefix());
+    }
+
 
     private OpenSearchSinkConfig generateOpenSearchSinkConfig(String pipelineName) throws IOException {
         final File configurationFile = new File(getClass().getClassLoader().getResource(OPEN_SEARCH_SINK_CONFIGURATIONS).getFile());

data-prepper-plugins/opensearch/src/test/resources/open-search-sink-configurations.yaml

@@ -8,6 +8,12 @@ valid-sink:
         sts_role_arn: "arn:aws:iam::123456789012:role/test-role"
         region: "us-east-2"
         serverless: true
+valid-sink-with-path-prefix:
+  sink:
+    opensearch:
+      hosts: [ "http://localhost:9200" ]
+      index: "no-more-plugin-setting"
+      path_prefix: "/os"
 invalid-action:
   sink:
     opensearch: