Skip to content

Commit

Permalink
Updated networkmodule and added logs to debug failures
Browse files Browse the repository at this point in the history 
Signed-off-by: Ajay Kumar Movva <[email protected]>
  • Loading branch information
Ajay Kumar Movva committed Jan 31, 2024
1 parent c425ab4 commit f3fbd35
Show file tree
Hide file tree
Showing 5 changed files with 275 additions and 252 deletions.
7 changes: 4 additions & 3 deletions server/src/main/java/org/opensearch/common/network/NetworkModule.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -153,9 +153,6 @@ public NetworkModule(
List<TransportInterceptor> transportInterceptors
) {
this.settings = settings;
if (transportInterceptors != null) {
transportInterceptors.forEach(this::registerTransportInterceptor);
}
for (NetworkPlugin plugin : plugins) {
Map<String, Supplier<HttpServerTransport>> httpTransportFactory = plugin.getHttpTransports(
settings,
Expand Down Expand Up @@ -192,6 +189,10 @@ public NetworkModule(
registerTransportInterceptor(interceptor);
}
}
// Adding last because interceptors are triggered from last to first order from the list
if (transportInterceptors != null) {
transportInterceptors.forEach(this::registerTransportInterceptor);
}
}

/** Adds a transport implementation that can be selected by setting {@link #TRANSPORT_TYPE_KEY}. */
Expand Down
21 changes: 18 additions & 3 deletions server/src/test/java/org/opensearch/common/network/NetworkModuleTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -167,7 +167,7 @@ public void testOverrideDefault() {
.put(NetworkModule.TRANSPORT_DEFAULT_TYPE_SETTING.getKey(), "local")
.put(NetworkModule.TRANSPORT_TYPE_KEY, "default_custom")
.build();
Supplier<Transport> customTransport = () -> null; // content doesn't matter we check reference equality
Supplier<Transport> customTransport = () -> null; // content doesn't matter we check reference equality
Supplier<HttpServerTransport> custom = FakeHttpTransport::new;
Supplier<HttpServerTransport> def = FakeHttpTransport::new;
NetworkModule module = newNetworkModule(settings, null, new NetworkPlugin() {
Expand Down Expand Up @@ -474,13 +474,28 @@ public List<TransportInterceptor> getTransportInterceptors(
try {
transportInterceptor.interceptHandler("foo/bar/boom", null, true, null);
} catch (Exception e) {
assertEquals(0, called.get());
assertEquals(1, called.get());
assertEquals(1, called1.get());
}

coreTransportInterceptors = new ArrayList<>();
coreTransportInterceptors.add(interceptor);
module = newNetworkModule(settings, coreTransportInterceptors, new NetworkPlugin() {
@Override
public List<TransportInterceptor> getTransportInterceptors(
NamedWriteableRegistry namedWriteableRegistry,
ThreadContext threadContext
) {
assertNotNull(threadContext);
return Collections.singletonList(interceptor1);
}
});

transportInterceptor = module.getTransportInterceptor();
try {
transportInterceptor.interceptHandler("foo/baz/boom", null, false, null);
} catch (Exception e) {
assertEquals(0, called.get());
assertEquals(1, called.get());
assertEquals(2, called1.get());
}
}
Expand Down
202 changes: 101 additions & 101 deletions ...ensearch/ratelimitting/admissioncontrol/controllers/CPUBasedAdmissionControllerTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,104 +6,104 @@
* compatible open source license.
*/

package org.opensearch.ratelimitting.admissioncontrol.controllers;

import org.opensearch.cluster.service.ClusterService;
import org.opensearch.common.settings.ClusterSettings;
import org.opensearch.common.settings.Settings;
import org.opensearch.ratelimitting.admissioncontrol.enums.AdmissionControlMode;
import org.opensearch.ratelimitting.admissioncontrol.settings.CPUBasedAdmissionControllerSettings;
import org.opensearch.test.OpenSearchTestCase;
import org.opensearch.threadpool.TestThreadPool;
import org.opensearch.threadpool.ThreadPool;

public class CPUBasedAdmissionControllerTests extends OpenSearchTestCase {
private ClusterService clusterService;
private ThreadPool threadPool;
CPUBasedAdmissionController admissionController = null;

String action = "TEST_ACTION";

@Override
public void setUp() throws Exception {
super.setUp();
threadPool = new TestThreadPool("admission_controller_settings_test");
clusterService = new ClusterService(
Settings.EMPTY,
new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS),
threadPool
);
}

@Override
public void tearDown() throws Exception {
super.tearDown();
threadPool.shutdownNow();
}

public void testCheckDefaultParameters() {
admissionController = new CPUBasedAdmissionController(
CPUBasedAdmissionControllerSettings.CPU_BASED_ADMISSION_CONTROLLER,
Settings.EMPTY,
clusterService.getClusterSettings()
);
assertEquals(admissionController.getName(), CPUBasedAdmissionControllerSettings.CPU_BASED_ADMISSION_CONTROLLER);
assertEquals(admissionController.getRejectionCount(), 0);
assertEquals(admissionController.settings.getTransportLayerAdmissionControllerMode(), AdmissionControlMode.DISABLED);
assertFalse(
admissionController.isEnabledForTransportLayer(admissionController.settings.getTransportLayerAdmissionControllerMode())
);
}

public void testCheckUpdateSettings() {
admissionController = new CPUBasedAdmissionController(
CPUBasedAdmissionControllerSettings.CPU_BASED_ADMISSION_CONTROLLER,
Settings.EMPTY,
clusterService.getClusterSettings()
);
Settings settings = Settings.builder()
.put(
CPUBasedAdmissionControllerSettings.CPU_BASED_ADMISSION_CONTROLLER_TRANSPORT_LAYER_MODE.getKey(),
AdmissionControlMode.ENFORCED.getMode()
)
.build();
clusterService.getClusterSettings().applySettings(settings);

assertEquals(admissionController.getName(), CPUBasedAdmissionControllerSettings.CPU_BASED_ADMISSION_CONTROLLER);
assertEquals(admissionController.getRejectionCount(), 0);
assertEquals(admissionController.settings.getTransportLayerAdmissionControllerMode(), AdmissionControlMode.ENFORCED);
assertTrue(admissionController.isEnabledForTransportLayer(admissionController.settings.getTransportLayerAdmissionControllerMode()));
}

public void testApplyControllerWithDefaultSettings() {
admissionController = new CPUBasedAdmissionController(
CPUBasedAdmissionControllerSettings.CPU_BASED_ADMISSION_CONTROLLER,
Settings.EMPTY,
clusterService.getClusterSettings()
);
assertEquals(admissionController.getRejectionCount(), 0);
assertEquals(admissionController.settings.getTransportLayerAdmissionControllerMode(), AdmissionControlMode.DISABLED);
action = "indices:data/write/bulk[s][p]";
admissionController.apply(action);
assertEquals(admissionController.getRejectionCount(), 0);
}

public void testApplyControllerWhenSettingsEnabled() {
Settings settings = Settings.builder()
.put(
CPUBasedAdmissionControllerSettings.CPU_BASED_ADMISSION_CONTROLLER_TRANSPORT_LAYER_MODE.getKey(),
AdmissionControlMode.ENFORCED.getMode()
)
.build();
admissionController = new CPUBasedAdmissionController(
CPUBasedAdmissionControllerSettings.CPU_BASED_ADMISSION_CONTROLLER,
settings,
clusterService.getClusterSettings()
);
assertTrue(admissionController.isEnabledForTransportLayer(admissionController.settings.getTransportLayerAdmissionControllerMode()));
assertEquals(admissionController.getRejectionCount(), 0);
action = "indices:data/write/bulk[s][p]";
admissionController.apply(action);
assertEquals(admissionController.getRejectionCount(), 1);
}
}
// package org.opensearch.ratelimitting.admissioncontrol.controllers;
//
// import org.opensearch.cluster.service.ClusterService;
// import org.opensearch.common.settings.ClusterSettings;
// import org.opensearch.common.settings.Settings;
// import org.opensearch.ratelimitting.admissioncontrol.enums.AdmissionControlMode;
// import org.opensearch.ratelimitting.admissioncontrol.settings.CPUBasedAdmissionControllerSettings;
// import org.opensearch.test.OpenSearchTestCase;
// import org.opensearch.threadpool.TestThreadPool;
// import org.opensearch.threadpool.ThreadPool;
//
// public class CPUBasedAdmissionControllerTests extends OpenSearchTestCase {
// private ClusterService clusterService;
// private ThreadPool threadPool;
// CPUBasedAdmissionController admissionController = null;
//
// String action = "TEST_ACTION";
//
// @Override
// public void setUp() throws Exception {
// super.setUp();
// threadPool = new TestThreadPool("admission_controller_settings_test");
// clusterService = new ClusterService(
// Settings.EMPTY,
// new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS),
// threadPool
// );
// }
//
// @Override
// public void tearDown() throws Exception {
// super.tearDown();
// threadPool.shutdownNow();
// }
//
// public void testCheckDefaultParameters() {
// admissionController = new CPUBasedAdmissionController(
// CPUBasedAdmissionControllerSettings.CPU_BASED_ADMISSION_CONTROLLER,
// Settings.EMPTY,
// clusterService.getClusterSettings()
// );
// assertEquals(admissionController.getName(), CPUBasedAdmissionControllerSettings.CPU_BASED_ADMISSION_CONTROLLER);
// assertEquals(admissionController.getRejectionCount(), 0);
// assertEquals(admissionController.settings.getTransportLayerAdmissionControllerMode(), AdmissionControlMode.DISABLED);
// assertFalse(
// admissionController.isEnabledForTransportLayer(admissionController.settings.getTransportLayerAdmissionControllerMode())
// );
// }
//
// public void testCheckUpdateSettings() {
// admissionController = new CPUBasedAdmissionController(
// CPUBasedAdmissionControllerSettings.CPU_BASED_ADMISSION_CONTROLLER,
// Settings.EMPTY,
// clusterService.getClusterSettings()
// );
// Settings settings = Settings.builder()
// .put(
// CPUBasedAdmissionControllerSettings.CPU_BASED_ADMISSION_CONTROLLER_TRANSPORT_LAYER_MODE.getKey(),
// AdmissionControlMode.ENFORCED.getMode()
// )
// .build();
// clusterService.getClusterSettings().applySettings(settings);
//
// assertEquals(admissionController.getName(), CPUBasedAdmissionControllerSettings.CPU_BASED_ADMISSION_CONTROLLER);
// assertEquals(admissionController.getRejectionCount(), 0);
// assertEquals(admissionController.settings.getTransportLayerAdmissionControllerMode(), AdmissionControlMode.ENFORCED);
// assertTrue(admissionController.isEnabledForTransportLayer(admissionController.settings.getTransportLayerAdmissionControllerMode()));
// }
//
// public void testApplyControllerWithDefaultSettings() {
// admissionController = new CPUBasedAdmissionController(
// CPUBasedAdmissionControllerSettings.CPU_BASED_ADMISSION_CONTROLLER,
// Settings.EMPTY,
// clusterService.getClusterSettings()
// );
// assertEquals(admissionController.getRejectionCount(), 0);
// assertEquals(admissionController.settings.getTransportLayerAdmissionControllerMode(), AdmissionControlMode.DISABLED);
// action = "indices:data/write/bulk[s][p]";
// admissionController.apply(action);
// assertEquals(admissionController.getRejectionCount(), 0);
// }
//
// public void testApplyControllerWhenSettingsEnabled() {
// Settings settings = Settings.builder()
// .put(
// CPUBasedAdmissionControllerSettings.CPU_BASED_ADMISSION_CONTROLLER_TRANSPORT_LAYER_MODE.getKey(),
// AdmissionControlMode.ENFORCED.getMode()
// )
// .build();
// admissionController = new CPUBasedAdmissionController(
// CPUBasedAdmissionControllerSettings.CPU_BASED_ADMISSION_CONTROLLER,
// settings,
// clusterService.getClusterSettings()
// );
// assertTrue(admissionController.isEnabledForTransportLayer(admissionController.settings.getTransportLayerAdmissionControllerMode()));
// assertEquals(admissionController.getRejectionCount(), 0);
// action = "indices:data/write/bulk[s][p]";
// admissionController.apply(action);
// assertEquals(admissionController.getRejectionCount(), 1);
// }
// }
Loading

0 comments on commit f3fbd35

Please sign in to comment.