Refactor auxiliary transport port settings framework.

The motivation of this change is to allow for multiple configured auxiliary transports. As such newAuxTransports() now returns a list of enabled transports and each AuxTransport implements it's own 'aux.transport.type' and 'aux.transport.<type>.ports' setting. Since Security.java initializes previous to Node.java during bootstrap socket binding permissions are granted based on 'aux.transport.<type>.ports' for each enabled 'aux.transport.type', falling back to a default if no ports are specified. Signed-off-by: Finn Carroll <[email protected]>
finnegancarroll · Dec 17, 2024 · 3ac47a6 · 3ac47a6
1 parent e1879ea
commit 3ac47a6
Show file tree

Hide file tree

Showing 10 changed files with 162 additions and 175 deletions.
diff --git a/plugins/transport-grpc/src/main/java/org/opensearch/transport/grpc/GrpcModulePlugin.java b/plugins/transport-grpc/src/main/java/org/opensearch/transport/grpc/GrpcModulePlugin.java
@@ -7,13 +7,11 @@
  */
 package org.opensearch.transport.grpc;
 
-import org.opensearch.common.lifecycle.LifecycleComponent;
 import org.opensearch.common.network.NetworkService;
 import org.opensearch.common.settings.ClusterSettings;
 import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.util.FeatureFlags;
-import org.opensearch.common.util.concurrent.OpenSearchExecutors;
 import org.opensearch.core.indices.breaker.CircuitBreakerService;
 import org.opensearch.plugins.NetworkPlugin;
 import org.opensearch.plugins.Plugin;
@@ -25,25 +23,21 @@
 import java.util.Map;
 import java.util.function.Supplier;
 
-import static org.opensearch.auxiliary.AuxTransportSettings.SETTING_AUX_BIND_HOST;
-import static org.opensearch.auxiliary.AuxTransportSettings.SETTING_AUX_HOST;
-import static org.opensearch.auxiliary.AuxTransportSettings.SETTING_AUX_PUBLISH_HOST;
+import static org.opensearch.transport.grpc.Netty4GrpcServerTransport.GRPC_TRANSPORT_SETTING_KEY;
+import static org.opensearch.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_BIND_HOST;
+import static org.opensearch.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_HOST;
+import static org.opensearch.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_PORTS;
+import static org.opensearch.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_PUBLISH_HOST;
+import static org.opensearch.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_PUBLISH_PORT;
+import static org.opensearch.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_WORKER_COUNT;
 
 /**
  * Main class for the gRPC plugin
  */
 public final class GrpcModulePlugin extends Plugin implements NetworkPlugin {
-    public static final String GRPC_TRANSPORT_NAME = "grpc-transport";
-
-    public static final Setting<Integer> SETTING_GRPC_WORKER_COUNT = new Setting<>(
-        "grpc.netty.worker_count",
-        (s) -> Integer.toString(OpenSearchExecutors.allocatedProcessors(s)),
-        (s) -> Setting.parseInt(s, 1, "grpc.netty.worker_count"),
-        Setting.Property.NodeScope
-    );
 
     @Override
-    public Map<String, Supplier<LifecycleComponent>> getAuxTransports(
+    public Map<String, Supplier<AuxTransport>> getAuxTransports(
         Settings settings,
         ThreadPool threadPool,
         CircuitBreakerService circuitBreakerService,
@@ -55,13 +49,20 @@ public Map<String, Supplier<LifecycleComponent>> getAuxTransports(
             throw new IllegalArgumentException("transport-grpc is experimental and feature flag must be enabled before use");
         }
         return Collections.singletonMap(
-            GRPC_TRANSPORT_NAME,
+            GRPC_TRANSPORT_SETTING_KEY,
             () -> new Netty4GrpcServerTransport(settings, Collections.emptyList(), networkService)
         );
     }
 
     @Override
     public List<Setting<?>> getSettings() {
-        return List.of(SETTING_AUX_HOST, SETTING_AUX_PUBLISH_HOST, SETTING_AUX_BIND_HOST, SETTING_GRPC_WORKER_COUNT);
+        return List.of(
+            SETTING_GRPC_PORTS,
+            SETTING_GRPC_HOST,
+            SETTING_GRPC_PUBLISH_HOST,
+            SETTING_GRPC_BIND_HOST,
+            SETTING_GRPC_WORKER_COUNT,
+            SETTING_GRPC_PUBLISH_PORT
+        );
     }
 }
diff --git a/...transport-grpc/src/main/java/org/opensearch/transport/grpc/Netty4GrpcServerTransport.java b/...transport-grpc/src/main/java/org/opensearch/transport/grpc/Netty4GrpcServerTransport.java
@@ -10,14 +10,15 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.opensearch.common.lifecycle.AbstractLifecycleComponent;
-import org.opensearch.common.lifecycle.LifecycleComponent;
 import org.opensearch.common.network.NetworkService;
+import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.transport.PortsRange;
+import org.opensearch.common.util.concurrent.OpenSearchExecutors;
 import org.opensearch.core.common.Strings;
 import org.opensearch.core.common.transport.BoundTransportAddress;
 import org.opensearch.core.common.transport.TransportAddress;
+import org.opensearch.plugins.NetworkPlugin;
 import org.opensearch.transport.BindTransportException;
 
 import java.io.IOException;
@@ -30,6 +31,7 @@
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
 
 import io.grpc.BindableService;
 import io.grpc.InsecureServerCredentials;
@@ -41,16 +43,48 @@
 import io.grpc.protobuf.services.HealthStatusManager;
 import io.grpc.protobuf.services.ProtoReflectionService;
 
-import static org.opensearch.auxiliary.AuxTransportSettings.SETTING_AUX_BIND_HOST;
-import static org.opensearch.auxiliary.AuxTransportSettings.SETTING_AUX_PORT;
-import static org.opensearch.auxiliary.AuxTransportSettings.SETTING_AUX_PUBLISH_HOST;
-import static org.opensearch.auxiliary.AuxTransportSettings.SETTING_AUX_PUBLISH_PORT;
+import static java.util.Collections.emptyList;
 import static org.opensearch.common.network.NetworkService.resolvePublishPort;
+import static org.opensearch.common.settings.Setting.intSetting;
+import static org.opensearch.common.settings.Setting.listSetting;
 import static org.opensearch.common.util.concurrent.OpenSearchExecutors.daemonThreadFactory;
-import static org.opensearch.transport.grpc.GrpcModulePlugin.SETTING_GRPC_WORKER_COUNT;
 
-public class Netty4GrpcServerTransport extends AbstractLifecycleComponent implements LifecycleComponent {
+public class Netty4GrpcServerTransport extends NetworkPlugin.AuxTransport {
     private static final Logger logger = LogManager.getLogger(Netty4GrpcServerTransport.class);
+    public static final String GRPC_TRANSPORT_SETTING_KEY = "grpc-transport";
+    public static final Setting<PortsRange> SETTING_GRPC_PORTS = AUX_TRANSPORT_PORTS.getConcreteSettingForNamespace(
+        GRPC_TRANSPORT_SETTING_KEY
+    );
+
+    public static final Setting<Integer> SETTING_GRPC_WORKER_COUNT = new Setting<>(
+        "grpc.netty.worker_count",
+        (s) -> Integer.toString(OpenSearchExecutors.allocatedProcessors(s)),
+        (s) -> Setting.parseInt(s, 1, "grpc.netty.worker_count"),
+        Setting.Property.NodeScope
+    );
+
+    public static final Setting<Integer> SETTING_GRPC_PUBLISH_PORT = intSetting("grpc.publish_port", -1, -1, Setting.Property.NodeScope);
+
+    public static final Setting<List<String>> SETTING_GRPC_HOST = listSetting(
+        "grpc.host",
+        emptyList(),
+        Function.identity(),
+        Setting.Property.NodeScope
+    );
+
+    public static final Setting<List<String>> SETTING_GRPC_PUBLISH_HOST = listSetting(
+        "grpc.publish_host",
+        SETTING_GRPC_HOST,
+        Function.identity(),
+        Setting.Property.NodeScope
+    );
+
+    public static final Setting<List<String>> SETTING_GRPC_BIND_HOST = listSetting(
+        "grpc.bind_host",
+        SETTING_GRPC_HOST,
+        Function.identity(),
+        Setting.Property.NodeScope
+    );
 
     private final Settings settings;
     private final NetworkService networkService;
@@ -69,16 +103,16 @@ public Netty4GrpcServerTransport(Settings settings, List<BindableService> servic
         this.services = Objects.requireNonNull(services);
         this.networkService = Objects.requireNonNull(networkService);
 
-        final List<String> httpBindHost = SETTING_AUX_BIND_HOST.get(settings);
+        final List<String> httpBindHost = SETTING_GRPC_BIND_HOST.get(settings);
         this.bindHosts = (httpBindHost.isEmpty() ? NetworkService.GLOBAL_NETWORK_BIND_HOST_SETTING.get(settings) : httpBindHost).toArray(
             Strings.EMPTY_ARRAY
         );
 
-        final List<String> httpPublishHost = SETTING_AUX_PUBLISH_HOST.get(settings);
+        final List<String> httpPublishHost = SETTING_GRPC_PUBLISH_HOST.get(settings);
         this.publishHosts = (httpPublishHost.isEmpty() ? NetworkService.GLOBAL_NETWORK_PUBLISH_HOST_SETTING.get(settings) : httpPublishHost)
             .toArray(Strings.EMPTY_ARRAY);
 
-        this.port = SETTING_AUX_PORT.get(settings);
+        this.port = SETTING_GRPC_PORTS.get(settings);
         this.nettyEventLoopThreads = SETTING_GRPC_WORKER_COUNT.get(settings);
     }
 
@@ -151,7 +185,7 @@ private void bindServer() {
             throw new BindTransportException("Failed to resolve publish address", e);
         }
 
-        final int publishPort = resolvePublishPort(SETTING_AUX_PUBLISH_PORT.get(settings), boundAddresses, publishInetAddress);
+        final int publishPort = resolvePublishPort(SETTING_GRPC_PUBLISH_PORT.get(settings), boundAddresses, publishInetAddress);
         if (publishPort < 0) {
             throw new BindTransportException(
                 "Failed to auto-resolve grpc publish port, multiple bound addresses "
@@ -160,9 +194,9 @@ private void bindServer() {
                     + publishInetAddress
                     + "). "
                     + "Please specify a unique port by setting "
-                    + SETTING_AUX_PORT.getKey()
+                    + SETTING_GRPC_PORTS.getKey()
                     + " or "
-                    + SETTING_AUX_PUBLISH_PORT.getKey()
+                    + SETTING_GRPC_PUBLISH_PORT.getKey()
             );
         }
 

diff --git a/...port-grpc/src/test/java/org/opensearch/transport/grpc/Netty4GrpcServerTransportTests.java b/...port-grpc/src/test/java/org/opensearch/transport/grpc/Netty4GrpcServerTransportTests.java
@@ -8,7 +8,6 @@
 
 package org.opensearch.transport.grpc;
 
-import org.opensearch.auxiliary.AuxTransportSettings;
 import org.opensearch.common.network.NetworkService;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.test.OpenSearchTestCase;
@@ -45,6 +44,6 @@ public void test() {
     }
 
     private static Settings createSettings() {
-        return Settings.builder().put(AuxTransportSettings.SETTING_AUX_PORT.getKey(), getPortRange()).build();
+        return Settings.builder().put(Netty4GrpcServerTransport.SETTING_GRPC_PORTS.getKey(), getPortRange()).build();
     }
 }
diff --git a/server/src/main/java/org/opensearch/auxiliary/AuxTransportSettings.java b/server/src/main/java/org/opensearch/auxiliary/AuxTransportSettings.java
diff --git a/server/src/main/java/org/opensearch/auxiliary/package-info.java b/server/src/main/java/org/opensearch/auxiliary/package-info.java
diff --git a/server/src/main/java/org/opensearch/bootstrap/Security.java b/server/src/main/java/org/opensearch/bootstrap/Security.java
@@ -32,11 +32,12 @@
 
 package org.opensearch.bootstrap;
 
-import org.opensearch.auxiliary.AuxTransportSettings;
 import org.opensearch.cli.Command;
 import org.opensearch.common.SuppressForbidden;
 import org.opensearch.common.io.PathUtils;
+import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.common.transport.PortsRange;
 import org.opensearch.env.Environment;
 import org.opensearch.http.HttpTransportSettings;
 import org.opensearch.plugins.PluginInfo;
@@ -72,7 +73,9 @@
 
 import static org.opensearch.bootstrap.FilePermissionUtils.addDirectoryPath;
 import static org.opensearch.bootstrap.FilePermissionUtils.addSingleFilePath;
-import static org.opensearch.common.network.NetworkModule.AUX_TRANSPORT_TYPE_SETTING;
+import static org.opensearch.plugins.NetworkPlugin.AuxTransport.AUX_PORT_DEFAULTS;
+import static org.opensearch.plugins.NetworkPlugin.AuxTransport.AUX_TRANSPORT_PORTS;
+import static org.opensearch.plugins.NetworkPlugin.AuxTransport.AUX_TRANSPORT_TYPES_SETTING;
 
 /**
  * Initializes SecurityManager with necessary permissions.
@@ -420,18 +423,26 @@ private static void addSocketPermissionForHttp(final Permissions policy, final S
     }
 
     /**
-     * Add dynamic {@link SocketPermission} based on auxiliary transport settings {@link AuxTransportSettings}.
-     * Socket permissions are not provided if no auxiliary transport is selected.
+     * Add dynamic {@link SocketPermission} based on AffixSetting AUX_TRANSPORT_PORTS.
+     * If an auxiliary transport type is enabled but has no corresponding port range setting fall back to AUX_PORT_DEFAULTS.
      *
      * @param policy the {@link Permissions} instance to apply the dynamic {@link SocketPermission}s to.
      * @param settings the {@link Settings} instance to read the gRPC settings from
      */
     private static void addSocketPermissionForAux(final Permissions policy, final Settings settings) {
-        if (!AUX_TRANSPORT_TYPE_SETTING.exists(settings)) {
-            return;
+        Set<PortsRange> portsRanges = new HashSet<>();
+        for (String auxType : AUX_TRANSPORT_TYPES_SETTING.get(settings)) {
+            Setting<PortsRange> auxTypePortSettings = AUX_TRANSPORT_PORTS.getConcreteSettingForNamespace(auxType);
+            if (auxTypePortSettings.exists(settings)) {
+                portsRanges.add(auxTypePortSettings.get(settings));
+            } else {
+                portsRanges.add(new PortsRange(AUX_PORT_DEFAULTS));
+            }
+        }
+
+        for (PortsRange portRange : portsRanges) {
+            addSocketPermissionForPortRange(policy, portRange.getPortRangeString());
         }
-        final String auxRange = AuxTransportSettings.SETTING_AUX_PORT.get(settings).getPortRangeString();
-        addSocketPermissionForPortRange(policy, auxRange);
     }
 
     /**