Merge branch 'main' of github.com:opensearch-project/OpenSearch into primary-rebalacing

Author: alchemist51

CHANGELOG.md

@@ -107,16 +107,19 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add a counter to node stat api to track shard going from idle to non-idle ([#12768](https://github.com/opensearch-project/OpenSearch/pull/12768))
 - Allow setting KEYSTORE_PASSWORD through env variable ([#12865](https://github.com/opensearch-project/OpenSearch/pull/12865))
 - [Concurrent Segment Search] Perform buildAggregation concurrently and support Composite Aggregations ([#12697](https://github.com/opensearch-project/OpenSearch/pull/12697))
+- [Concurrent Segment Search] Disable concurrent segment search for system indices and throttled requests ([#12954](https://github.com/opensearch-project/OpenSearch/pull/12954))
 - Add cluster primary balance contraint for rebalancing with buffer ([#12656](https://github.com/opensearch-project/OpenSearch/pull/12656))
 
 ### Dependencies
 - Bump `org.apache.commons:commons-configuration2` from 2.10.0 to 2.10.1 ([#12896](https://github.com/opensearch-project/OpenSearch/pull/12896))
 - Bump `asm` from 9.6 to 9.7 ([#12908](https://github.com/opensearch-project/OpenSearch/pull/12908))
 - Bump `net.minidev:json-smart` from 2.5.0 to 2.5.1 ([#12893](https://github.com/opensearch-project/OpenSearch/pull/12893))
 - Bump `netty` from 4.1.107.Final to 4.1.108.Final ([#12924](https://github.com/opensearch-project/OpenSearch/pull/12924))
+- Bump `commons-io:commons-io` from 2.15.1 to 2.16.0 ([#12996](https://github.com/opensearch-project/OpenSearch/pull/12996))
 
 ### Changed
 - [BWC and API enforcement] Enforcing the presence of API annotations at build time ([#12872](https://github.com/opensearch-project/OpenSearch/pull/12872))
+- Improve built-in secure transports support ([#12907](https://github.com/opensearch-project/OpenSearch/pull/12907))
 
 ### Deprecated
 

modules/transport-netty4/src/main/java/org/opensearch/http/netty4/ssl/SecureNetty4HttpServerTransport.java

@@ -37,19 +37,27 @@
 import org.opensearch.core.xcontent.NamedXContentRegistry;
 import org.opensearch.http.HttpChannel;
 import org.opensearch.http.HttpHandlingSettings;
+import org.opensearch.http.HttpServerTransport;
 import org.opensearch.http.netty4.Netty4HttpChannel;
 import org.opensearch.http.netty4.Netty4HttpServerTransport;
-import org.opensearch.plugins.SecureTransportSettingsProvider;
+import org.opensearch.plugins.SecureHttpTransportSettingsProvider;
+import org.opensearch.plugins.TransportExceptionHandler;
 import org.opensearch.telemetry.tracing.Tracer;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.SharedGroupFactory;
+import org.opensearch.transport.TransportAdapterProvider;
 import org.opensearch.transport.netty4.ssl.SslUtils;
 
 import javax.net.ssl.SSLEngine;
 
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.handler.codec.DecoderException;
 import io.netty.handler.ssl.ApplicationProtocolNames;
 import io.netty.handler.ssl.ApplicationProtocolNegotiationHandler;
@@ -59,9 +67,14 @@
  * @see <a href="https://github.com/opensearch-project/security/blob/d526c9f6c2a438c14db8b413148204510b9fe2e2/src/main/java/org/opensearch/security/ssl/http/netty/SecuritySSLNettyHttpServerTransport.java">SecuritySSLNettyHttpServerTransport</a>
  */
 public class SecureNetty4HttpServerTransport extends Netty4HttpServerTransport {
+    public static final String REQUEST_HEADER_VERIFIER = "HeaderVerifier";
+    public static final String REQUEST_DECOMPRESSOR = "RequestDecompressor";
+
     private static final Logger logger = LogManager.getLogger(SecureNetty4HttpServerTransport.class);
-    private final SecureTransportSettingsProvider secureTransportSettingsProvider;
-    private final SecureTransportSettingsProvider.ServerExceptionHandler exceptionHandler;
+    private final SecureHttpTransportSettingsProvider secureHttpTransportSettingsProvider;
+    private final TransportExceptionHandler exceptionHandler;
+    private final ChannelInboundHandlerAdapter headerVerifier;
+    private final TransportAdapterProvider<HttpServerTransport> decompressorProvider;
 
     public SecureNetty4HttpServerTransport(
         final Settings settings,
@@ -72,7 +85,7 @@ public SecureNetty4HttpServerTransport(
         final Dispatcher dispatcher,
         final ClusterSettings clusterSettings,
         final SharedGroupFactory sharedGroupFactory,
-        final SecureTransportSettingsProvider secureTransportSettingsProvider,
+        final SecureHttpTransportSettingsProvider secureHttpTransportSettingsProvider,
         final Tracer tracer
     ) {
         super(
@@ -86,9 +99,45 @@ public SecureNetty4HttpServerTransport(
             sharedGroupFactory,
             tracer
         );
-        this.secureTransportSettingsProvider = secureTransportSettingsProvider;
-        this.exceptionHandler = secureTransportSettingsProvider.buildHttpServerExceptionHandler(settings, this)
-            .orElse(SecureTransportSettingsProvider.ServerExceptionHandler.NOOP);
+
+        this.secureHttpTransportSettingsProvider = secureHttpTransportSettingsProvider;
+        this.exceptionHandler = secureHttpTransportSettingsProvider.buildHttpServerExceptionHandler(settings, this)
+            .orElse(TransportExceptionHandler.NOOP);
+
+        final List<ChannelInboundHandlerAdapter> headerVerifiers = secureHttpTransportSettingsProvider.getHttpTransportAdapterProviders(
+            settings
+        )
+            .stream()
+            .filter(p -> REQUEST_HEADER_VERIFIER.equalsIgnoreCase(p.name()))
+            .map(p -> p.create(settings, this, ChannelInboundHandlerAdapter.class))
+            .filter(Optional::isPresent)
+            .map(Optional::get)
+            .collect(Collectors.toList());
+
+        if (headerVerifiers.size() > 1) {
+            throw new IllegalArgumentException("Cannot have more than one header verifier configured, supplied " + headerVerifiers.size());
+        }
+
+        final Optional<TransportAdapterProvider<HttpServerTransport>> decompressorProviderOpt = secureHttpTransportSettingsProvider
+            .getHttpTransportAdapterProviders(settings)
+            .stream()
+            .filter(p -> REQUEST_DECOMPRESSOR.equalsIgnoreCase(p.name()))
+            .findFirst();
+        // There could be multiple request decompressor providers configured, using the first one
+        decompressorProviderOpt.ifPresent(p -> logger.debug("Using request decompressor provider: {}", p));
+
+        this.headerVerifier = headerVerifiers.isEmpty() ? null : headerVerifiers.get(0);
+        this.decompressorProvider = decompressorProviderOpt.orElseGet(() -> new TransportAdapterProvider<HttpServerTransport>() {
+            @Override
+            public String name() {
+                return REQUEST_DECOMPRESSOR;
+            }
+
+            @Override
+            public <C> Optional<C> create(Settings settings, HttpServerTransport transport, Class<C> adapterClass) {
+                return Optional.empty();
+            }
+        });
     }
 
     @Override
@@ -152,7 +201,7 @@ protected SslHttpChannelHandler(final Netty4HttpServerTransport transport, final
         protected void initChannel(Channel ch) throws Exception {
             super.initChannel(ch);
 
-            final SSLEngine sslEngine = secureTransportSettingsProvider.buildSecureHttpServerEngine(
+            final SSLEngine sslEngine = secureHttpTransportSettingsProvider.buildSecureHttpServerEngine(
                 settings,
                 SecureNetty4HttpServerTransport.this
             ).orElseGet(SslUtils::createDefaultServerSSLEngine);
@@ -166,4 +215,17 @@ protected void configurePipeline(Channel ch) {
             ch.pipeline().addLast(new Http2OrHttpHandler());
         }
     }
+
+    protected ChannelInboundHandlerAdapter createHeaderVerifier() {
+        if (headerVerifier != null) {
+            return headerVerifier;
+        } else {
+            return super.createHeaderVerifier();
+        }
+    }
+
+    @Override
+    protected ChannelInboundHandlerAdapter createDecompressor() {
+        return decompressorProvider.create(settings, this, ChannelInboundHandlerAdapter.class).orElseGet(super::createDecompressor);
+    }
 }

modules/transport-netty4/src/main/java/org/opensearch/transport/Netty4ModulePlugin.java

@@ -49,6 +49,7 @@
 import org.opensearch.http.netty4.ssl.SecureNetty4HttpServerTransport;
 import org.opensearch.plugins.NetworkPlugin;
 import org.opensearch.plugins.Plugin;
+import org.opensearch.plugins.SecureHttpTransportSettingsProvider;
 import org.opensearch.plugins.SecureTransportSettingsProvider;
 import org.opensearch.telemetry.tracing.Tracer;
 import org.opensearch.threadpool.ThreadPool;
@@ -160,7 +161,7 @@ public Map<String, Supplier<HttpServerTransport>> getSecureHttpTransports(
         NetworkService networkService,
         HttpServerTransport.Dispatcher dispatcher,
         ClusterSettings clusterSettings,
-        SecureTransportSettingsProvider secureTransportSettingsProvider,
+        SecureHttpTransportSettingsProvider secureHttpTransportSettingsProvider,
         Tracer tracer
     ) {
         return Collections.singletonMap(
@@ -174,7 +175,7 @@ public Map<String, Supplier<HttpServerTransport>> getSecureHttpTransports(
                 dispatcher,
                 clusterSettings,
                 getSharedGroupFactory(settings),
-                secureTransportSettingsProvider,
+                secureHttpTransportSettingsProvider,
                 tracer
             )
         );

modules/transport-netty4/src/main/java/org/opensearch/transport/netty4/ssl/SecureNetty4Transport.java

@@ -42,6 +42,7 @@
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.core.indices.breaker.CircuitBreakerService;
 import org.opensearch.plugins.SecureTransportSettingsProvider;
+import org.opensearch.plugins.TransportExceptionHandler;
 import org.opensearch.telemetry.tracing.Tracer;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.SharedGroupFactory;
@@ -72,7 +73,7 @@ public class SecureNetty4Transport extends Netty4Transport {
 
     private static final Logger logger = LogManager.getLogger(SecureNetty4Transport.class);
     private final SecureTransportSettingsProvider secureTransportSettingsProvider;
-    private final SecureTransportSettingsProvider.ServerExceptionHandler exceptionHandler;
+    private final TransportExceptionHandler exceptionHandler;
 
     public SecureNetty4Transport(
         final Settings settings,
@@ -100,7 +101,7 @@ public SecureNetty4Transport(
 
         this.secureTransportSettingsProvider = secureTransportSettingsProvider;
         this.exceptionHandler = secureTransportSettingsProvider.buildServerTransportExceptionHandler(settings, this)
-            .orElse(SecureTransportSettingsProvider.ServerExceptionHandler.NOOP);
+            .orElse(TransportExceptionHandler.NOOP);
     }
 
     @Override