Moved exclude_identification_keys to under peer_forwarder config

Signed-off-by: Krishna Kondaka <[email protected]>
opensearch-project · Oct 29, 2024 · 28e4ef7 · 28e4ef7
1 parent 286e822
commit 28e4ef7
Show file tree

Hide file tree

Showing 8 changed files with 29 additions and 28 deletions.
diff --git a/...repper-core/src/main/java/org/opensearch/dataprepper/core/parser/PipelineTransformer.java b/...repper-core/src/main/java/org/opensearch/dataprepper/core/parser/PipelineTransformer.java
@@ -204,8 +204,9 @@ private void buildPipelineFromConfiguration(
                         if (!processors.isEmpty() && processors.get(0) instanceof RequiresPeerForwarding) {
                             return PeerForwardingProcessorDecorator.decorateProcessors(
                                     processors, peerForwarderProvider, pipelineName, processorComponentList.get(0).getName(),
-				    dataPrepperConfiguration.getExcludeIdentificationKeys(),
-				    pipelineConfiguration.getWorkers()
+                                    dataPrepperConfiguration.getPeerForwarderConfiguration() != null ?
+                                        dataPrepperConfiguration.getPeerForwarderConfiguration().getExcludeIdentificationKeys() : null,
+                                    pipelineConfiguration.getWorkers()
                             );
                         }
                         return processors;

diff --git a/.../src/main/java/org/opensearch/dataprepper/core/parser/model/DataPrepperConfiguration.java b/.../src/main/java/org/opensearch/dataprepper/core/parser/model/DataPrepperConfiguration.java
@@ -22,12 +22,10 @@
 
 import java.time.Duration;
 import java.util.Collections;
-import java.util.stream.Collectors;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.Objects;
 
 /**
@@ -58,7 +56,6 @@ public class DataPrepperConfiguration implements ExtensionsConfiguration, EventC
     private Duration processorShutdownTimeout;
     private Duration sinkShutdownTimeout;
     private PipelineExtensions pipelineExtensions;
-    private List<Set<String>> excludeIdentificationKeys;
 
     public static final DataPrepperConfiguration DEFAULT_CONFIG = new DataPrepperConfiguration();
 
@@ -76,8 +73,6 @@ public DataPrepperConfiguration(
             @JsonProperty("private_key_password")
             @JsonAlias("privateKeyPassword")
             final String privateKeyPassword,
-            @JsonProperty("exclude_identification_keys")
-            final List<Set<String>> excludeIdentificationKeys,
             @JsonProperty("server_port")
             @JsonAlias("serverPort")
             final String serverPort,
@@ -132,7 +127,6 @@ public DataPrepperConfiguration(
             throw new IllegalArgumentException("sinkShutdownTimeout must be non-negative.");
         }
         this.pipelineExtensions = pipelineExtensions;
-        this.excludeIdentificationKeys = excludeIdentificationKeys;
     }
 
     public int getServerPort() {
@@ -143,13 +137,6 @@ public boolean ssl() {
         return ssl;
     }
 
-    public Set<Set<String>> getExcludeIdentificationKeys() {
-        if (excludeIdentificationKeys == null) {
-            return null;
-        }
-        return excludeIdentificationKeys.stream().collect(Collectors.toSet());
-    }
-
     public String getKeyStoreFilePath() {
         return keyStoreFilePath;
     }

diff --git a/...c/main/java/org/opensearch/dataprepper/core/peerforwarder/PeerForwarderConfiguration.java b/...c/main/java/org/opensearch/dataprepper/core/peerforwarder/PeerForwarderConfiguration.java
@@ -14,8 +14,10 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.stream.Collectors;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 /**
  * Class to hold configuration for Core Peer Forwarder in {@link DataPrepperConfiguration},
@@ -65,6 +67,7 @@ public class PeerForwarderConfiguration {
     private Integer forwardingBatchQueueDepth = 1;
     private Duration forwardingBatchTimeout = DEFAULT_FORWARDING_BATCH_TIMEOUT;
     private boolean binaryCodec = true;
+    private List<Set<String>> excludeIdentificationKeys;
 
     public PeerForwarderConfiguration() {}
 
@@ -76,6 +79,7 @@ public PeerForwarderConfiguration (
             @JsonProperty("server_thread_count") final Integer serverThreadCount,
             @JsonProperty("max_connection_count") final Integer maxConnectionCount,
             @JsonProperty("max_pending_requests") final Integer maxPendingRequests,
+            @JsonProperty("exclude_identification_keys") final List<Set<String>> excludeIdentificationKeys,
             @JsonProperty("ssl") final Boolean ssl,
             @JsonProperty("ssl_certificate_file") final String sslCertificateFile,
             @JsonProperty("ssl_key_file") final String sslKeyFile,
@@ -139,6 +143,7 @@ public PeerForwarderConfiguration (
         setBinaryCodec(binaryCodec == null || binaryCodec);
         checkForCertAndKeyFileInS3();
         validateSslAndAuthentication();
+        this.excludeIdentificationKeys = excludeIdentificationKeys;
     }
 
     public int getServerPort() {
@@ -169,6 +174,13 @@ public boolean isSsl() {
         return ssl;
     }
 
+    public Set<Set<String>> getExcludeIdentificationKeys() {
+        if (excludeIdentificationKeys == null) {
+            return null;
+        }
+        return excludeIdentificationKeys.stream().collect(Collectors.toSet());
+    }
+
     public String getSslCertificateFile() {
         return sslCertificateFile;
     }

diff --git a/...test/java/org/opensearch/dataprepper/core/parser/model/DataPrepperConfigurationTests.java b/...test/java/org/opensearch/dataprepper/core/parser/model/DataPrepperConfigurationTests.java
@@ -27,7 +27,6 @@
 import java.io.IOException;
 import java.time.Duration;
 import java.util.Map;
-import java.util.Set;
 
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.notNullValue;
@@ -60,15 +59,13 @@ public void testSomeDefaultConfig() throws IOException {
                 makeConfig(TestDataProvider.VALID_DATA_PREPPER_SOME_DEFAULT_CONFIG_FILE);
         Assert.assertEquals(DataPrepperConfiguration.DEFAULT_CONFIG.getServerPort(), dataPrepperConfiguration.getServerPort());
         Assert.assertNull(dataPrepperConfiguration.getPipelineExtensions());
-        Assert.assertNull(dataPrepperConfiguration.getExcludeIdentificationKeys());
     }
 
     @Test
     public void testDefaultMetricsRegistry() {
         final DataPrepperConfiguration dataPrepperConfiguration = DataPrepperConfiguration.DEFAULT_CONFIG;
         assertThat(dataPrepperConfiguration.getMetricRegistryTypes().size(), Matchers.equalTo(1));
         assertThat(dataPrepperConfiguration.getMetricRegistryTypes(), Matchers.hasItem(MetricRegistryType.Prometheus));
-        Assert.assertNull(dataPrepperConfiguration.getExcludeIdentificationKeys());
     }
 
     @Test
@@ -88,13 +85,6 @@ public void testMultipleMetricsRegistry() throws IOException {
         assertThat(dataPrepperConfiguration.getMetricRegistryTypes(), Matchers.hasItem(MetricRegistryType.CloudWatch));
     }
 
-    @Test
-    public void testExcludeIdentificationKeys() throws IOException {
-        final DataPrepperConfiguration dataPrepperConfiguration =
-                makeConfig(TestDataProvider.VALID_DATA_PREPPER_CONFIG_FILE);
-        Assert.assertEquals(Set.of(Set.of("key1"), Set.of("key2", "key3")), dataPrepperConfiguration.getExcludeIdentificationKeys());
-    }
-
     @Test
     void testConfigurationWithHttpBasic() throws IOException {
         final DataPrepperConfiguration dataPrepperConfiguration =

diff --git a/...st/java/org/opensearch/dataprepper/core/peerforwarder/PeerForwarderConfigurationTest.java b/...st/java/org/opensearch/dataprepper/core/peerforwarder/PeerForwarderConfigurationTest.java
@@ -22,6 +22,7 @@
 import java.time.Duration;
 import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
+import java.util.Set;
 
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -123,6 +124,14 @@ void testValidPeerForwarderConfig_with_InsecureTls() throws IOException {
 
         assertThat(peerForwarderConfiguration.isSsl(), equalTo(true));
         assertThat(peerForwarderConfiguration.isSslDisableVerification(), equalTo(true));
+        assertThat(peerForwarderConfiguration.getExcludeIdentificationKeys(), equalTo(null));
+    }
+
+    @Test
+    public void testExcludeIdentificationKeys() throws IOException {
+        final PeerForwarderConfiguration peerForwarderConfiguration =
+                makeConfig("src/test/resources/valid_peer_forwarder_config.yml");
+        assertThat(Set.of(Set.of("key1"), Set.of("key2", "key3")), equalTo(peerForwarderConfiguration.getExcludeIdentificationKeys()));
     }
 
     @Test
@@ -131,6 +140,7 @@ void testValidPeerForwarderConfig_with_FingerprintTls() throws IOException {
 
         assertThat(peerForwarderConfiguration.isSsl(), equalTo(true));
         assertThat(peerForwarderConfiguration.isSslFingerprintVerificationOnly(), equalTo(true));
+        assertThat(peerForwarderConfiguration.getExcludeIdentificationKeys(), equalTo(null));
     }
 
     @Test
@@ -191,4 +201,4 @@ void test_cert_paths_with_ssl() throws IOException {
     void invalid_InvalidPeerForwarderConfig_test(final String filePath) {
         assertThrows(ValueInstantiationException.class, () -> makeConfig(filePath));
     }
-}
+}
diff --git a/...test/java/org/opensearch/dataprepper/core/peerforwarder/PeerForwarder_ClientServerIT.java b/...test/java/org/opensearch/dataprepper/core/peerforwarder/PeerForwarder_ClientServerIT.java
@@ -519,6 +519,7 @@ private PeerForwarderConfiguration createConfiguration(
                 200,
                 500,
                 1024,
+                null,
                 ssl,
                 sslCertificateFile,
                 sslKeyFile,

diff --git a/data-prepper-core/src/test/resources/valid_data_prepper_config.yml b/data-prepper-core/src/test/resources/valid_data_prepper_config.yml
@@ -1,3 +1,2 @@
 server_port: 5678
-exclude_identification_keys: [ ["key1"], ["key2", "key3"]]
 ssl: false
diff --git a/data-prepper-core/src/test/resources/valid_peer_forwarder_config.yml b/data-prepper-core/src/test/resources/valid_peer_forwarder_config.yml
@@ -4,6 +4,7 @@ client_timeout: 50
 server_thread_count: 100
 max_connection_count: 100
 max_pending_requests: 512
+exclude_identification_keys: [ ["key1"], ["key2", "key3"] ]
 ssl: false
 use_acm_certificate_for_ssl: false
 discovery_mode: static
@@ -15,4 +16,4 @@ failed_forwarding_requests_local_write_timeout: 15
 forwarding_batch_size: 2500
 forwarding_batch_queue_depth: 3
 forwarding_batch_timeout: 5s
-binary_codec: false
+binary_codec: false