Incorporated review comments changes for opensearch-project#1985,open…

…search-project#2264. Signed-off-by:rajeshLovesToCode <[email protected]>
rajeshLovesToCode · May 12, 2023 · 2afc0c8 · 2afc0c8
1 parent 28ac950
commit 2afc0c8
Show file tree

Hide file tree

Showing 8 changed files with 54 additions and 103 deletions.
diff --git a/data-prepper-plugins/opensearch-source/build.gradle b/data-prepper-plugins/opensearch-source/build.gradle
@@ -8,6 +8,7 @@ dependencies {
     implementation 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.14.2'
     implementation 'software.amazon.awssdk:s3'
     implementation 'software.amazon.awssdk:sts'
+    implementation project(path: ':data-prepper-core')
     testImplementation 'com.fasterxml.jackson.dataformat:jackson-dataformat-yaml'
 }
 

diff --git a/...a/org/opensearch/dataprepper/plugins/source/opensearch/OpenSearchSourceConfiguration.java b/...a/org/opensearch/dataprepper/plugins/source/opensearch/OpenSearchSourceConfiguration.java
@@ -5,19 +5,31 @@
 package org.opensearch.dataprepper.plugins.source.opensearch;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
+import jakarta.validation.constraints.NotNull;
 import org.opensearch.dataprepper.plugins.source.opensearch.configuration.AwsAuthenticationConfiguration;
 import org.opensearch.dataprepper.plugins.source.opensearch.configuration.ConnectionConfiguration;
 import org.opensearch.dataprepper.plugins.source.opensearch.configuration.IndexParametersConfiguration;
 import org.opensearch.dataprepper.plugins.source.opensearch.configuration.QueryParameterConfiguration;
 import org.opensearch.dataprepper.plugins.source.opensearch.configuration.SchedulingParameterConfiguration;
-import org.opensearch.dataprepper.plugins.source.opensearch.configuration.RetryConfiguration;
 import org.opensearch.dataprepper.plugins.source.opensearch.configuration.SearchConfiguration;
 
-
-import java.util.Map;
+import java.util.List;
 
 public class OpenSearchSourceConfiguration {
 
+    @JsonProperty("max_retries")
+    private Integer maxRetries;
+
+    @NotNull
+    @JsonProperty("hosts")
+    private List<String> hosts;
+
+    @JsonProperty("username")
+    private String username;
+
+    @JsonProperty("password")
+    private String password;
+
     @JsonProperty("connection")
     private ConnectionConfiguration connectionConfiguration;
 
@@ -36,10 +48,21 @@ public class OpenSearchSourceConfiguration {
     @JsonProperty("search_options")
     private SearchConfiguration searchConfiguration;
 
-    @JsonProperty("retry")
-    private RetryConfiguration retryConfiguration;
+    public Integer getMaxRetries() {
+        return maxRetries;
+    }
+
+    public List<String> getHosts() {
+        return hosts;
+    }
+
+    public String getUsername() {
+        return username;
+    }
 
-    private Map<String,String> indexNames;
+    public String getPassword() {
+        return password;
+    }
 
     public ConnectionConfiguration getConnectionConfiguration() {
         return connectionConfiguration;
@@ -65,11 +88,4 @@ public SearchConfiguration getSearchConfiguration() {
         return searchConfiguration;
     }
 
-    public Map<String, String> getIndexNames() {
-        return indexNames;
-    }
-
-    public RetryConfiguration getRetryConfiguration() {
-        return retryConfiguration;
-    }
 }
diff --git a/...ensearch/dataprepper/plugins/source/opensearch/configuration/ConnectionConfiguration.java b/...ensearch/dataprepper/plugins/source/opensearch/configuration/ConnectionConfiguration.java
@@ -5,22 +5,10 @@
 package org.opensearch.dataprepper.plugins.source.opensearch.configuration;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
-import jakarta.validation.constraints.NotNull;
 import java.nio.file.Path;
-import java.util.List;
 
 public class ConnectionConfiguration {
 
-  @NotNull
-  @JsonProperty("hosts")
-  private List<String> hosts;
-
-  @JsonProperty("username")
-  private String username;
-
-  @JsonProperty("password")
-  private String password;
-
   @JsonProperty("cert")
   private Path certPath;
 
@@ -33,18 +21,6 @@ public class ConnectionConfiguration {
   @JsonProperty("insecure")
   private boolean insecure;
 
-  public List<String> getHosts() {
-    return hosts;
-  }
-
-  public String getUsername() {
-    return username;
-  }
-
-  public String getPassword() {
-    return password;
-  }
-
   public Path getCertPath() {
     return certPath;
   }
@@ -57,4 +33,7 @@ public Integer getConnectTimeout() {
     return connectTimeout;
   }
 
+  public boolean isInsecure() {
+    return insecure;
+  }
 }
diff --git a/...rg/opensearch/dataprepper/plugins/source/opensearch/configuration/RetryConfiguration.java b/...rg/opensearch/dataprepper/plugins/source/opensearch/configuration/RetryConfiguration.java
diff --git a/...dataprepper/plugins/source/opensearch/configuration/SchedulingParameterConfiguration.java b/...dataprepper/plugins/source/opensearch/configuration/SchedulingParameterConfiguration.java
@@ -8,25 +8,23 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
-import com.fasterxml.jackson.datatype.jsr310.deser.DurationDeserializer;
 import com.fasterxml.jackson.datatype.jsr310.deser.LocalDateTimeDeserializer;
-import com.fasterxml.jackson.datatype.jsr310.ser.DurationSerializer;
 import com.fasterxml.jackson.datatype.jsr310.ser.LocalDateTimeSerializer;
 import jakarta.validation.constraints.Min;
+import org.opensearch.dataprepper.parser.DataPrepperDurationDeserializer;
 
 import java.time.Duration;
 import java.time.LocalDateTime;
 
 public class SchedulingParameterConfiguration {
 
-    @JsonSerialize(using = DurationSerializer.class)
-    @JsonDeserialize(using = DurationDeserializer.class)
     @JsonProperty("rate")
+    @JsonDeserialize(using = DataPrepperDurationDeserializer.class)
     private Duration rate;
 
     @Min(1)
     @JsonProperty("job_count")
-    private int jobCount;
+    private int jobCount = 1;
 
     @JsonSerialize(using = LocalDateTimeSerializer.class)
     @JsonDeserialize(using = LocalDateTimeDeserializer.class)

diff --git a/...g/opensearch/dataprepper/plugins/source/opensearch/OpenSearchSourceConfigurationTest.java b/...g/opensearch/dataprepper/plugins/source/opensearch/OpenSearchSourceConfigurationTest.java
@@ -31,10 +31,12 @@ public class OpenSearchSourceConfigurationTest {
     @Test
     public void open_search_source_config_values_test() throws JsonProcessingException {
 
-        final String sourceConfigurationYaml = "connection:\n" +
-                "  hosts: [\"http://localhost:9200\"]\n" +
-                "  username: test\n" +
-                "  password: test\n" +
+        final String sourceConfigurationYaml = "max_retries: 5\n" +
+                "hosts: [\"http://localhost:9200\"]\n" +
+                "username: test\n" +
+                "password: test\n" +
+                "connection:\n" +
+                "  insecure: true\n" +
                 "  cert: \"cert\"\n" +
                 "indices:\n" +
                 "  include:\n" +
@@ -52,25 +54,29 @@ public void open_search_source_config_values_test() throws JsonProcessingExcepti
                 "  batch_size: 1000\n" +
                 "  expand_wildcards: \"open\"\n" +
                 "  sorting:\n" +
-                "retry: \n" +
-                "  max_retries: 3";
+                "   - sort_key: name\n" +
+                "     order: desc";
         final OpenSearchSourceConfiguration sourceConfiguration = objectMapper.readValue(sourceConfigurationYaml, OpenSearchSourceConfiguration.class);
         final ConnectionConfiguration connectionConfig = sourceConfiguration.getConnectionConfiguration();
         final SearchConfiguration searchConfiguration = sourceConfiguration.getSearchConfiguration();
         final AwsAuthenticationConfiguration awsAuthenticationOptions = sourceConfiguration.getAwsAuthenticationOptions();
         final SchedulingParameterConfiguration schedulingParameterConfiguration = sourceConfiguration.getSchedulingParameterConfiguration();
+
+        assertThat(sourceConfiguration.getMaxRetries(),equalTo(5));
         assertThat(awsAuthenticationOptions.getAwsRegion(),equalTo(Region.US_EAST_1));
-        assertThat(connectionConfig.getHosts().get(0),equalTo("http://localhost:9200"));
-        assertThat(connectionConfig.getUsername(),equalTo("test"));
-        assertThat(connectionConfig.getPassword(),equalTo("test"));
+        assertThat(sourceConfiguration.getHosts().get(0),equalTo("http://localhost:9200"));
+        assertThat(sourceConfiguration.getUsername(),equalTo("test"));
+        assertThat(sourceConfiguration.getPassword(),equalTo("test"));
         assertThat(connectionConfig.getCertPath(),equalTo(Path.of("cert")));
+        assertThat(connectionConfig.isInsecure(),equalTo(true));
         assertThat(searchConfiguration.getExpandWildcards(),equalTo(WildCardOptions.OPEN));
         assertThat(searchConfiguration.getBatchSize(),equalTo(1000));
         assertThat(sourceConfiguration.getQueryParameterConfiguration().getFields(),equalTo(List.of("test_variable : test_value")));
         assertThat(schedulingParameterConfiguration.getRate(),equalTo(Duration.parse("P2DT3H4M")));
         assertThat(schedulingParameterConfiguration.getJobCount(),equalTo(3));
         assertThat(schedulingParameterConfiguration.getStartTime(),equalTo(LocalDateTime.parse("2023-05-05T18:00:00")));
         assertThat(sourceConfiguration.getIndexParametersConfiguration().getInclude().get(0),equalTo("shakespeare"));
-        assertThat(sourceConfiguration.getRetryConfiguration().getMaxRetries(),equalTo(3));
+        assertThat(searchConfiguration.getSorting().get(0).getSortKey(),equalTo("name"));
+        assertThat(searchConfiguration.getSorting().get(0).getOrder(),equalTo("desc"));
     }
 }
diff --git a/...arch/dataprepper/plugins/source/opensearch/configuration/ConnectionConfigurationTest.java b/...arch/dataprepper/plugins/source/opensearch/configuration/ConnectionConfigurationTest.java
@@ -20,19 +20,15 @@ public class ConnectionConfigurationTest {
     @Test
     public void connection_configuration_values_test() throws JsonProcessingException {
 
-        final String connectionYaml = "  hosts: [\"http://localhost:9200\"]\n" +
-                "  username: test\n" +
-                "  password: test\n" +
+        final String connectionYaml =
                 "  cert: \"cert\"\n" +
                 "  insecure: true\n" +
                 "  socket_timeout: 500\n" +
                 "  connection_timeout: 500";
         final ConnectionConfiguration connectionConfig = objectMapper.readValue(connectionYaml, ConnectionConfiguration.class);
-        assertThat(connectionConfig.getHosts().get(0),equalTo("http://localhost:9200"));
-        assertThat(connectionConfig.getUsername(),equalTo("test"));
-        assertThat(connectionConfig.getPassword(),equalTo("test"));
         assertThat(connectionConfig.getCertPath(),equalTo(Path.of("cert")));
         assertThat(connectionConfig.getSocketTimeout(),equalTo(500));
         assertThat(connectionConfig.getConnectTimeout(),equalTo(500));
+        assertThat(connectionConfig.isInsecure(),equalTo(true));
     }
 }
diff --git a/...pensearch/dataprepper/plugins/source/opensearch/configuration/RetryConfigurationTest.java b/...pensearch/dataprepper/plugins/source/opensearch/configuration/RetryConfigurationTest.java