[Feature/extensions] Adding support for Extension Create Components [environment settings and add setting update consumers]

src/main/java/org/opensearch/sdk/ExtensionsRunner.java

@@ -24,11 +24,16 @@
 import org.opensearch.extensions.rest.RestExecuteOnExtensionResponse;
 import org.opensearch.common.network.NetworkModule;
 import org.opensearch.common.network.NetworkService;
+import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.util.PageCacheRecycler;
 import org.opensearch.extensions.ExtensionBooleanResponse;
 import org.opensearch.discovery.InitializeExtensionsRequest;
 import org.opensearch.discovery.InitializeExtensionsResponse;
+import org.opensearch.extensions.DiscoveryExtension;
+import org.opensearch.extensions.EnvironmentSettingsRequest;
+import org.opensearch.extensions.AddSettingsUpdateConsumerRequest;
+import org.opensearch.extensions.UpdateSettingsRequest;
 import org.opensearch.extensions.ExtensionRequest;
 import org.opensearch.extensions.ExtensionsOrchestrator;
 import org.opensearch.index.IndicesModuleRequest;
@@ -41,8 +46,10 @@
 import org.opensearch.rest.RestResponse;
 import org.opensearch.transport.netty4.Netty4Transport;
 import org.opensearch.transport.SharedGroupFactory;
+import org.opensearch.sdk.handlers.AddSettingsUpdateConsumerResponseHandler;
 import org.opensearch.sdk.handlers.ClusterSettingsResponseHandler;
 import org.opensearch.sdk.handlers.ClusterStateResponseHandler;
+import org.opensearch.sdk.handlers.EnvironmentSettingsResponseHandler;
 import org.opensearch.sdk.handlers.LocalNodeResponseHandler;
 import org.opensearch.sdk.handlers.RegisterRestActionsResponseHandler;
 import org.opensearch.search.SearchModule;
@@ -53,15 +60,21 @@
 import org.opensearch.transport.TransportService;
 import org.opensearch.transport.TransportSettings;
 import org.opensearch.transport.TransportResponse;
+import org.opensearch.common.settings.WriteableSetting;
+import org.opensearch.common.unit.ByteSizeValue;
+import org.opensearch.common.unit.TimeValue;
 
 import java.io.File;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import java.util.function.Consumer;
 
 import static java.util.Collections.emptySet;
 import static org.opensearch.common.UUIDs.randomBase64UUID;
@@ -78,6 +91,7 @@ public class ExtensionsRunner {
 
     private String uniqueId;
     private DiscoveryNode opensearchNode;
+    private DiscoveryExtension extensionNode;
     private TransportService extensionTransportService = null;
     private ExtensionRestPathRegistry extensionRestPathRegistry = new ExtensionRestPathRegistry();
 
@@ -90,6 +104,7 @@ public class ExtensionsRunner {
      * https://github.com/opensearch-project/opensearch-sdk-java/issues/119
      */
     private TransportActions transportActions = new TransportActions(new HashMap<>());
+    private Map<Setting<?>, Consumer<?>> settingUpdateConsumers = new HashMap<>();
 
     /**
      * Instantiates a new Extensions Runner using test settings.
@@ -152,6 +167,10 @@ private void setOpensearchNode(DiscoveryNode opensearchNode) {
         this.opensearchNode = opensearchNode;
     }
 
+    private void setExtensionNode(DiscoveryExtension extensionNode) {
+        this.extensionNode = extensionNode;
+    }
+
     DiscoveryNode getOpensearchNode() {
         return opensearchNode;
     }
@@ -172,6 +191,7 @@ InitializeExtensionsResponse handleExtensionInitRequest(InitializeExtensionsRequ
         } finally {
             // After sending successful response to initialization, send the REST API
             setOpensearchNode(opensearchNode);
+            setExtensionNode(extensionInitRequest.getExtension());
             extensionTransportService.connectToNode(opensearchNode);
             sendRegisterRestActionsRequest(extensionTransportService);
             transportActions.sendRegisterTransportActionsRequest(extensionTransportService, opensearchNode);
@@ -222,6 +242,66 @@ ExtensionBooleanResponse handleIndicesModuleNameRequest(IndicesModuleRequest ind
         return indicesModuleNameResponse;
     }
 
+    /**
+     * Handles a request to update a setting from OpenSearch.  Extensions must register their setting keys and consumers within the settingUpdateConsumer map
+     *
+     * @param updateSettingsRequest  The request to handle.
+     * @return A response acknowledging the request.
+     */
+    @SuppressWarnings("unchecked")
+    ExtensionBooleanResponse handleUpdateSettingsRequest(UpdateSettingsRequest updateSettingsRequest) {
+        logger.info("Registering UpdateSettingsRequest received from OpenSearch");
+
+        boolean settingUpdateStatus = true;
+
+        WriteableSetting.WriteableSettingGenericType settingType = updateSettingsRequest.getSettingType();
+        Setting<?> componentSetting = updateSettingsRequest.getComponentSetting();
+        Object data = updateSettingsRequest.getData();
+
+        // Setting updater in OpenSearch performs setting change validation, only need to cast the consumer to the corresponding type and
+        // invoke the consumer
+        try {
+            switch (settingType) {
+                case Boolean:
+                    Consumer<Boolean> boolConsumer = (Consumer<Boolean>) this.settingUpdateConsumers.get(componentSetting);
+                    boolConsumer.accept(Boolean.parseBoolean(data.toString()));
+                case Integer:
+                    Consumer<Integer> intConsumer = (Consumer<Integer>) this.settingUpdateConsumers.get(componentSetting);
+                    intConsumer.accept(Integer.parseInt(data.toString()));
+                case Long:
+                    Consumer<Long> longConsumer = (Consumer<Long>) this.settingUpdateConsumers.get(componentSetting);
+                    longConsumer.accept(Long.parseLong(data.toString()));
+                case Float:
+                    Consumer<Float> floatConsumer = (Consumer<Float>) this.settingUpdateConsumers.get(componentSetting);
+                    floatConsumer.accept(Float.parseFloat(data.toString()));
+                case Double:
+                    Consumer<Double> doubleConsumer = (Consumer<Double>) this.settingUpdateConsumers.get(componentSetting);
+                    doubleConsumer.accept(Double.parseDouble(data.toString()));
+                case String:
+                    Consumer<String> stringConsumer = (Consumer<String>) this.settingUpdateConsumers.get(componentSetting);
+                    stringConsumer.accept(data.toString());
+                case TimeValue:
+                    Consumer<TimeValue> timeValueConsumer = (Consumer<TimeValue>) this.settingUpdateConsumers.get(componentSetting);
+                    timeValueConsumer.accept(TimeValue.parseTimeValue(data.toString(), componentSetting.getKey()));
+                case ByteSizeValue:
+                    Consumer<ByteSizeValue> byteSizeValueConsumer = (Consumer<ByteSizeValue>) this.settingUpdateConsumers.get(
+                        componentSetting
+                    );
+                    byteSizeValueConsumer.accept(ByteSizeValue.parseBytesSizeValue(data.toString(), componentSetting.getKey()));
+                case Version:
+                    Consumer<Version> versionConsumer = (Consumer<Version>) this.settingUpdateConsumers.get(componentSetting);
+                    versionConsumer.accept((Version) data);
+                default:
+                    throw new UnsupportedOperationException("Setting Update Consumer type does not exist and is not handled here");
+            }
+        } catch (Exception e) {
+            logger.info(e.getMessage());
+            settingUpdateStatus = false;
+        }
+
+        return new ExtensionBooleanResponse(settingUpdateStatus);
+    }
+
     /**
      * Handles a request from OpenSearch to execute a REST request on the extension.
      *
@@ -392,6 +472,15 @@ public void startTransportService(TransportService transportService) {
             ((request, channel, task) -> channel.sendResponse(handleRestExecuteOnExtensionRequest(request)))
         );
 
+        transportService.registerRequestHandler(
+            ExtensionsOrchestrator.REQUEST_EXTENSION_UPDATE_SETTINGS,
+            ThreadPool.Names.GENERIC,
+            false,
+            false,
+            UpdateSettingsRequest::new,
+            ((request, channel, task) -> channel.sendResponse(handleUpdateSettingsRequest(request)))
+        );
+
     }
 
     /**
@@ -475,6 +564,74 @@ public void sendLocalNodeRequest(TransportService transportService) {
         }
     }
 
+    /**
+     * Requests the environment setting values from OpenSearch for the corresponding component settings. The result will be handled by a {@link EnvironmentSettingsResponseHandler}.
+     *
+     * @param componentSettings The component setting that correspond to the values provided by the environment settings
+     * @param transportService  The TransportService defining the connection to OpenSearch.
+     */
+    public void sendEnvironmentSettingsRequest(TransportService transportService, List<Setting<?>> componentSettings) {
+        logger.info("Sending Environment Settings request to OpenSearch");
+        EnvironmentSettingsResponseHandler environmentSettingsResponseHandler = new EnvironmentSettingsResponseHandler();
+        try {
+            transportService.sendRequest(
+                opensearchNode,
+                ExtensionsOrchestrator.REQUEST_EXTENSION_ENVIRONMENT_SETTINGS,
+                new EnvironmentSettingsRequest(componentSettings),
+                environmentSettingsResponseHandler
+            );
+        } catch (Exception e) {
+            logger.info("Failed to send Environment Settings request to OpenSearch", e);
+        }
+    }
+
+    /**
+     * Registers the component {@link Setting} and the corresponding consumer to the settingsUpdateConsumer map.
+     * Extensions will use this method instead of invoking ClusterSettings#addSettingsUpdateConsumer
+     *
+     * @param componentSetting The component setting associated with the consumer
+     * @param consumer  The setting update consumer associated with the component setting
+     */
+    public <T> void registerSettingUpdateConsumer(Setting<T> componentSetting, Consumer<T> consumer) {
+        this.settingUpdateConsumers.put(componentSetting, consumer);
+    }
+
+    /**
+     * Requests the environment setting values from OpenSearch for the corresponding component settings.
+     * This should be called after extension components have registered all their setting update consumers with the settingUpdateConsumers map.
+     * The result will be handled by a {@link EnvironmentSettingsResponseHandler}.
+     *
+     * @param transportService  The TransportService defining the connection to OpenSearch.
+     * @throws Exception if there are no setting update consumers registered within the settingUpdateConsumers map
+     */
+    public void sendAddSettingsUpdateConsumerRequest(TransportService transportService) throws Exception {
+        logger.info("Sending Add Settings Update Consumer request to OpenSearch");
+
+        // Determine if setting update consumers have been registered
+        if (this.settingUpdateConsumers.isEmpty()) {
+            throw new Exception("There are no setting update consumers registered");
+        } else {
+
+            // Extract registered settings from setting update consumer map
+            List<Setting<?>> componentSettings = new ArrayList<>(this.settingUpdateConsumers.size());
+            componentSettings.addAll(this.settingUpdateConsumers.keySet());
+
+            AddSettingsUpdateConsumerResponseHandler addSettingsUpdateConsumerResponseHandler =
+                new AddSettingsUpdateConsumerResponseHandler();
+            try {
+                transportService.sendRequest(
+                    opensearchNode,
+                    ExtensionsOrchestrator.REQUEST_EXTENSION_ADD_SETTINGS_UPDATE_CONSUMER,
+                    new AddSettingsUpdateConsumerRequest(this.extensionNode, componentSettings),
+                    addSettingsUpdateConsumerResponseHandler
+                );
+            } catch (Exception e) {
+                logger.info("Failed to send Add Settings Update Consumer request to OpenSearch", e);
+            }
+        }
+
+    }
+
     private Settings getSettings() {
         return settings;
     }

src/main/java/org/opensearch/sdk/handlers/AddSettingsUpdateConsumerResponseHandler.java

@@ -0,0 +1,46 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+package org.opensearch.sdk.handlers;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.extensions.ExtensionBooleanResponse;
+import org.opensearch.common.io.stream.StreamInput;
+import org.opensearch.sdk.ExtensionsRunner;
+import org.opensearch.threadpool.ThreadPool;
+import org.opensearch.transport.TransportException;
+import org.opensearch.transport.TransportResponseHandler;
+
+import java.io.IOException;
+
+/**
+ * This class handles the response from OpenSearch to a {@link ExtensionsRunner#sendAddSettingsUpdateConsumerRequest} call.
+ */
+public class AddSettingsUpdateConsumerResponseHandler implements TransportResponseHandler<ExtensionBooleanResponse> {
+    private static final Logger logger = LogManager.getLogger(AddSettingsUpdateConsumerResponseHandler.class);
+
+    @Override
+    public void handleResponse(ExtensionBooleanResponse response) {
+        logger.info("received {}", response);
+    }
+
+    @Override
+    public void handleException(TransportException exp) {
+        logger.info("AddSettingsUpdateConsumerResponse failed", exp);
+    }
+
+    @Override
+    public String executor() {
+        return ThreadPool.Names.GENERIC;
+    }
+
+    @Override
+    public ExtensionBooleanResponse read(StreamInput in) throws IOException {
+        return new ExtensionBooleanResponse(in);
+    }
+}

src/main/java/org/opensearch/sdk/handlers/EnvironmentSettingsResponseHandler.java

@@ -0,0 +1,46 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+package org.opensearch.sdk.handlers;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.env.EnvironmentSettingsResponse;
+import org.opensearch.common.io.stream.StreamInput;
+import org.opensearch.sdk.ExtensionsRunner;
+import org.opensearch.threadpool.ThreadPool;
+import org.opensearch.transport.TransportException;
+import org.opensearch.transport.TransportResponseHandler;
+
+import java.io.IOException;
+
+/**
+ * This class handles the response from OpenSearch to a {@link ExtensionsRunner#sendEnvironmentSettingsRequest} call.
+ */
+public class EnvironmentSettingsResponseHandler implements TransportResponseHandler<EnvironmentSettingsResponse> {
+    private static final Logger logger = LogManager.getLogger(EnvironmentSettingsResponseHandler.class);
+
+    @Override
+    public void handleResponse(EnvironmentSettingsResponse response) {
+        logger.info("received {}", response);
+    }
+
+    @Override
+    public void handleException(TransportException exp) {
+        logger.info("EnvironmentSettingsRequest failed", exp);
+    }
+
+    @Override
+    public String executor() {
+        return ThreadPool.Names.GENERIC;
+    }
+
+    @Override
+    public EnvironmentSettingsResponse read(StreamInput in) throws IOException {
+        return new EnvironmentSettingsResponse(in);
+    }
+}