Moved configuration reloading to a single dedicated thread (opensearch-project#5479)

Signed-off-by: Nils Bandener <[email protected]>
Signed-off-by: Nils Bandener <[email protected]>
Signed-off-by: Darshit Chanpura <[email protected]>
Co-authored-by: Darshit Chanpura <[email protected]>
Signed-off-by: Dennis Toepker <[email protected]>

Author: 2 people

CHANGELOG.md

@@ -9,6 +9,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Features
 
 ### Enhancements
+- Moved configuration reloading to dedicated thread to improve node stability  ([#5479](https://github.com/opensearch-project/security/pull/5479))
 - Makes resource settings dynamic ([#5677](https://github.com/opensearch-project/security/pull/5677))
 - [Resource Sharing] Allow multiple sharable resource types in single resource index ([#5713](https://github.com/opensearch-project/security/pull/5713))
 

src/main/java/org/opensearch/security/OpenSearchSecurityPlugin.java

@@ -1181,6 +1181,7 @@ public Collection<Object> createComponents(
         final XFFResolver xffResolver = new XFFResolver(threadPool);
         backendRegistry = new BackendRegistry(settings, adminDns, xffResolver, auditLog, threadPool, cih);
         backendRegistry.registerClusterSettingsChangeListener(clusterService.getClusterSettings());
+        cr.subscribeOnChange(configMap -> { backendRegistry.invalidateCache(); });
         tokenManager = new SecurityTokenManager(cs, threadPool, userService);
 
         final CompatConfig compatConfig = new CompatConfig(environment, transportPassiveAuthSetting);

src/main/java/org/opensearch/security/action/configupdate/TransportConfigUpdateAction.java

@@ -35,22 +35,22 @@
 
 import org.opensearch.action.FailedNodeException;
 import org.opensearch.action.support.ActionFilters;
-import org.opensearch.action.support.nodes.TransportNodesAction;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.inject.Inject;
 import org.opensearch.common.inject.Provider;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.StreamOutput;
 import org.opensearch.security.auth.BackendRegistry;
 import org.opensearch.security.configuration.ConfigurationRepository;
-import org.opensearch.security.securityconf.DynamicConfigFactory;
 import org.opensearch.security.securityconf.impl.CType;
+import org.opensearch.security.util.TransportNodesAsyncAction;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.TransportRequest;
 import org.opensearch.transport.TransportService;
 
-public class TransportConfigUpdateAction extends TransportNodesAction<
+public class TransportConfigUpdateAction extends TransportNodesAsyncAction<
     ConfigUpdateRequest,
     ConfigUpdateResponse,
     TransportConfigUpdateAction.NodeConfigUpdateRequest,
@@ -59,7 +59,6 @@ public class TransportConfigUpdateAction extends TransportNodesAction<
     protected Logger logger = LogManager.getLogger(getClass());
     private final Provider<BackendRegistry> backendRegistry;
     private final ConfigurationRepository configurationRepository;
-    private DynamicConfigFactory dynamicConfigFactory;
     private static final Set<CType<?>> SELECTIVE_VALIDATION_TYPES = Set.of(CType.INTERNALUSERS);
     // Note: While INTERNALUSERS is used as a marker, the cache invalidation
     // applies to all user types (internal, LDAP, etc.)
@@ -72,8 +71,7 @@ public TransportConfigUpdateAction(
         final TransportService transportService,
         final ConfigurationRepository configurationRepository,
         final ActionFilters actionFilters,
-        Provider<BackendRegistry> backendRegistry,
-        DynamicConfigFactory dynamicConfigFactory
+        Provider<BackendRegistry> backendRegistry
     ) {
         super(
             ConfigUpdateAction.NAME,
@@ -84,12 +82,12 @@ public TransportConfigUpdateAction(
             ConfigUpdateRequest::new,
             TransportConfigUpdateAction.NodeConfigUpdateRequest::new,
             ThreadPool.Names.MANAGEMENT,
+            ThreadPool.Names.SAME,
             ConfigUpdateNodeResponse.class
         );
 
         this.configurationRepository = configurationRepository;
         this.backendRegistry = backendRegistry;
-        this.dynamicConfigFactory = dynamicConfigFactory;
     }
 
     public static class NodeConfigUpdateRequest extends TransportRequest {
@@ -128,17 +126,29 @@ protected ConfigUpdateResponse newResponse(
     }
 
     @Override
-    protected ConfigUpdateNodeResponse nodeOperation(final NodeConfigUpdateRequest request) {
+    protected void nodeOperation(NodeConfigUpdateRequest request, ActionListener<ConfigUpdateNodeResponse> listener) {
         final var configupdateRequest = request.request;
         if (canHandleSelectively(configupdateRequest)) {
             backendRegistry.get().invalidateUserCache(configupdateRequest.getEntityNames());
+            listener.onResponse(new ConfigUpdateNodeResponse(clusterService.localNode(), configupdateRequest.getConfigTypes(), null));
         } else {
-            boolean didReload = configurationRepository.reloadConfiguration(CType.fromStringValues((configupdateRequest.getConfigTypes())));
-            if (didReload) {
-                backendRegistry.get().invalidateCache();
-            }
+            configurationRepository.reloadConfiguration(
+                CType.fromStringValues((configupdateRequest.getConfigTypes())),
+                new ActionListener<>() {
+                    @Override
+                    public void onResponse(ConfigurationRepository.ConfigReloadResponse configReloadResponse) {
+                        listener.onResponse(
+                            new ConfigUpdateNodeResponse(clusterService.localNode(), configupdateRequest.getConfigTypes(), null)
+                        );
+                    }
+
+                    @Override
+                    public void onFailure(Exception e) {
+                        listener.onFailure(e);
+                    }
+                }
+            );
         }
-        return new ConfigUpdateNodeResponse(clusterService.localNode(), configupdateRequest.getConfigTypes(), null);
     }
 
     private boolean canHandleSelectively(ConfigUpdateRequest request) {