TransportConfigUpdateAction does wait again for config reload, but no longer blocks thread

Signed-off-by: Nils Bandener <[email protected]>

Author: nibix

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

@@ -35,21 +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.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,
@@ -81,6 +82,7 @@ public TransportConfigUpdateAction(
             ConfigUpdateRequest::new,
             TransportConfigUpdateAction.NodeConfigUpdateRequest::new,
             ThreadPool.Names.MANAGEMENT,
+            ThreadPool.Names.SAME,
             ConfigUpdateNodeResponse.class
         );
 
@@ -124,14 +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 {
-            configurationRepository.reloadConfiguration(CType.fromStringValues((configupdateRequest.getConfigTypes())));
+            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) {

src/main/java/org/opensearch/security/configuration/ConfigurationRepository.java

@@ -531,9 +531,11 @@ public <T> SecurityDynamicConfiguration<T> getConfiguration(CType<T> configurati
      * updates into one.
 
      * @param configTypes the configuration types to be reloaded.
+     * @param listener an listener to be notified when the reload was finished. You can provide null if you do not want
+     *                 such a notification
      */
-    public void reloadConfiguration(Collection<CType<?>> configTypes) {
-        this.reloadThread.requestReload(configTypes);
+    public void reloadConfiguration(Collection<CType<?>> configTypes, ActionListener<ConfigReloadResponse> listener) {
+        this.reloadThread.requestReload(configTypes, listener);
     }
 
     /**
@@ -651,7 +653,7 @@ public void afterIndexShardStarted(IndexShard indexShard) {
                 threadPool.generic().execute(() -> {
                     if (isSecurityIndexRestoredFromSnapshot(clusterService, index, securityIndex)) {
                         LOGGER.info("Security index primary shard {} started - config reloading for snapshot restore", shardId);
-                        reloadConfiguration(CType.values());
+                        reloadConfiguration(CType.values(), null);
                     }
                 });
             }
@@ -689,6 +691,12 @@ static class ReloadThread {
          */
         private ImmutableSet<CType<?>> reloadRequestedFor = ImmutableSet.of();
 
+        /**
+         * Action listeners to be called when the reload was finished. We collect the action listeners here until
+         * the reload is actually in progress.
+         */
+        private List<ActionListener<ConfigReloadResponse>> reloadRequestedForActionListeners = new ArrayList<>();
+
         /**
          * The time we got the first currently queued reload request.
          */
@@ -708,12 +716,16 @@ static class ReloadThread {
          * Requests an async configuration reload for the given configuration types. Calling this method
          * will not wait for the configuration reload to complete.
          */
-        void requestReload(Collection<CType<?>> configurationTypes) {
+        void requestReload(Collection<CType<?>> configurationTypes, ActionListener<ConfigReloadResponse> actionListener) {
             synchronized (this.requestLock) {
                 if (!this.started) {
                     LOGGER.info("Cannot reload configuration yet, because the initialization process did not complete yet");
                 }
 
+                if (actionListener != null) {
+                    this.reloadRequestedForActionListeners.add(actionListener);
+                }
+
                 if (this.reloadRequestedFor.isEmpty()) {
                     LOGGER.debug("Configuration reload request received for {}; notifying update thread", configurationTypes);
                     this.reloadRequestedAt = Instant.now();
@@ -778,8 +790,9 @@ boolean queueIsEmpty() {
 
         private void run() {
             for (;;) {
+                ImmutableSet<CType<?>> localReloadRequestedFor;
+                List<ActionListener<ConfigReloadResponse>> localReloadRequestedForActionListeners = null;
                 try {
-                    ImmutableSet<CType<?>> localReloadRequestedFor;
 
                     synchronized (this.requestLock) {
                         this.reloadInProgressFor = ImmutableSet.of();
@@ -790,8 +803,8 @@ private void run() {
 
                         // We save here the requested configuration types in order to pass them to the updateFunction later
                         localReloadRequestedFor = this.reloadRequestedFor;
+                        localReloadRequestedForActionListeners = new ArrayList<>(this.reloadRequestedForActionListeners);
 
-                        // this.reloadRequestedAt != null: This means, that we got an update request
                         LOGGER.info(
                             "Performing configuration reload for request at {} on {}",
                             this.reloadRequestedAt,
@@ -802,14 +815,35 @@ private void run() {
                         // following update process will be already recognized again and queued.
                         this.reloadRequestedAt = null;
                         this.reloadRequestedFor = ImmutableSet.of();
+                        this.reloadRequestedForActionListeners.clear();
                         this.reloadInProgressFor = localReloadRequestedFor;
                     }
 
                     this.performFunction.accept(localReloadRequestedFor);
+                    for (ActionListener<ConfigReloadResponse> listener : localReloadRequestedForActionListeners) {
+                        listener.onResponse(new ConfigReloadResponse(localReloadRequestedFor));
+                    }
                 } catch (Exception e) {
                     LOGGER.error("Error in {}", this.thread.getName(), e);
+                    if (localReloadRequestedForActionListeners != null) {
+                        for (ActionListener<ConfigReloadResponse> listener : localReloadRequestedForActionListeners) {
+                            listener.onFailure(e);
+                        }
+                    }
                 }
             }
         }
     }
+
+    public static class ConfigReloadResponse {
+        private final Set<CType<?>> reloadedConfigTypes;
+
+        ConfigReloadResponse(Set<CType<?>> reloadedConfigTypes) {
+            this.reloadedConfigTypes = reloadedConfigTypes;
+        }
+
+        public Set<CType<?>> getReloadedConfigTypes() {
+            return reloadedConfigTypes;
+        }
+    }
 }