Enable TLS for Netty4GrpcServerTransport

CHANGELOG.md

@@ -23,6 +23,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - [Security Manager Replacement] Enhance Java Agent to intercept Runtime::halt ([#17757](https://github.com/opensearch-project/OpenSearch/pull/17757))
 - [Security Manager Replacement] Phase off SecurityManager usage in favor of Java Agent ([#17861](https://github.com/opensearch-project/OpenSearch/pull/17861))
 - Support AutoExpand for SearchReplica ([#17741](https://github.com/opensearch-project/OpenSearch/pull/17741))
+- Add TLS enabled SecureNetty4GrpcServerTransport ([#17796](https://github.com/opensearch-project/OpenSearch/pull/17796))
 - Implement fixed interval refresh task scheduling ([#17777](https://github.com/opensearch-project/OpenSearch/pull/17777))
 - [Tiered caching] Create a single cache manager for all the disk caches. ([#17513](https://github.com/opensearch-project/OpenSearch/pull/17513))
 - Add GRPC DocumentService and Bulk endpoint ([#17727](https://github.com/opensearch-project/OpenSearch/pull/17727))

plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/bootstrap/FlightService.java

@@ -134,6 +134,7 @@ public StreamManager getStreamManager() {
      * Retrieves the bound address of the FlightService.
      * @return The BoundTransportAddress instance.
      */
+    @Override
     public BoundTransportAddress getBoundAddress() {
         return serverComponents.getBoundAddress();
     }

plugins/transport-grpc/README.md

@@ -0,0 +1,42 @@
+# transport-grpc
+
+An auxiliary transport which runs in parallel to the REST API.
+The `transport-grpc` plugin initializes a new client/server transport implementing a gRPC protocol on Netty4.
+
+Enable this transport with:
+
+```
+setting 'aux.transport.types',                              '[experimental-transport-grpc]'
+setting 'aux.transport.experimental-transport-grpc.port',   '9400-9500' //optional
+```
+
+For the secure transport:
+
+```
+setting 'aux.transport.types',                                      '[experimental-secure-transport-grpc]'
+setting 'aux.transport.experimental-secure-transport-grpc.port',    '9400-9500' //optional
+```
+
+Other settings are agnostic as to the gRPC transport type:
+
+```
+setting 'grpc.publish_port',        '9400'
+setting 'grpc.host',                '["0.0.0.0"]'
+setting 'grpc.bind_host',           '["0.0.0.0", "::", "10.0.0.1"]'
+setting 'grpc.publish_host',        '["thisnode.example.com"]'
+setting 'grpc.netty.worker_count',  '2'
+```
+
+## Testing
+
+### Unit Tests
+
+```
+./gradlew :plugins:transport-grpc:test
+```
+
+### Integration Tests
+
+```
+./gradlew :plugins:transport-grpc:internalClusterTest
+```

plugins/transport-grpc/build.gradle

@@ -1,5 +1,3 @@
-import org.gradle.api.attributes.java.TargetJvmEnvironment
-
 /*
  * SPDX-License-Identifier: Apache-2.0
  *
@@ -8,13 +6,21 @@ import org.gradle.api.attributes.java.TargetJvmEnvironment
  * compatible open source license.
  */
 
+apply plugin: 'opensearch.testclusters'
 apply plugin: 'opensearch.internal-cluster-test'
 
 opensearchplugin {
   description = 'gRPC based transport implementation'
   classname = 'org.opensearch.plugin.transport.grpc.GrpcPlugin'
 }
 
+testClusters {
+  integTest {
+    plugin(project.path)
+    setting 'aux.transport.types', '[experimental-transport-grpc]'
+  }
+}
+
 dependencies {
   compileOnly "com.google.code.findbugs:jsr305:3.0.2"
   runtimeOnly "com.google.guava:guava:${versions.guava}"
@@ -30,6 +36,7 @@ dependencies {
   implementation "io.grpc:grpc-util:${versions.grpc}"
   implementation "io.perfmark:perfmark-api:0.26.0"
   implementation "org.opensearch:protobufs:0.2.0"
+  testImplementation project(':test:framework')
 }
 
 tasks.named("dependencyLicenses").configure {

plugins/transport-grpc/src/internalClusterTest/java/org/opensearch/plugin/transport/grpc/GrpcTransportIT.java → plugins/transport-grpc/src/internalClusterTest/java/org/opensearch/plugin/transport/grpc/Netty4GrpcServerTransportIT.java

@@ -8,35 +8,45 @@
 
 package org.opensearch.plugin.transport.grpc;
 
+import org.opensearch.action.admin.cluster.health.ClusterHealthResponse;
+import org.opensearch.cluster.health.ClusterHealthStatus;
 import org.opensearch.common.network.NetworkAddress;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.core.common.transport.TransportAddress;
+import org.opensearch.plugin.transport.grpc.ssl.NettyGrpcClient;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.test.OpenSearchIntegTestCase;
 
 import java.net.InetSocketAddress;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
+
+import io.grpc.health.v1.HealthCheckResponse;
 
 import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.GRPC_TRANSPORT_SETTING_KEY;
-import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_PORT;
 import static org.opensearch.plugins.NetworkPlugin.AuxTransport.AUX_TRANSPORT_TYPES_KEY;
 
-@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 2)
-public class GrpcTransportIT extends OpenSearchIntegTestCase {
+public class Netty4GrpcServerTransportIT extends OpenSearchIntegTestCase {
 
-    @Override
-    protected Collection<Class<? extends Plugin>> nodePlugins() {
-        return Collections.singletonList(GrpcPlugin.class);
+    private TransportAddress randomNetty4GrpcServerTransportAddr() {
+        List<TransportAddress> addresses = new ArrayList<>();
+        for (Netty4GrpcServerTransport transport : internalCluster().getInstances(Netty4GrpcServerTransport.class)) {
+            TransportAddress tAddr = new TransportAddress(transport.getBoundAddress().publishAddress().address());
+            addresses.add(tAddr);
+        }
+        return randomFrom(addresses);
     }
 
     @Override
     protected Settings nodeSettings(int nodeOrdinal) {
-        return Settings.builder()
-            .put(super.nodeSettings(nodeOrdinal))
-            .put(SETTING_GRPC_PORT.getKey(), "0")
-            .put(AUX_TRANSPORT_TYPES_KEY, GRPC_TRANSPORT_SETTING_KEY)
-            .build();
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal)).put(AUX_TRANSPORT_TYPES_KEY, GRPC_TRANSPORT_SETTING_KEY).build();
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return Collections.singleton(GrpcPlugin.class);
     }
 
     public void testGrpcTransportStarted() {
@@ -46,7 +56,7 @@ public void testGrpcTransportStarted() {
             assertNotNull("gRPC transport should be started on node " + nodeName, transport);
 
             // Verify that the transport is bound to an address
-            TransportAddress[] boundAddresses = transport.boundAddress().boundAddresses();
+            TransportAddress[] boundAddresses = transport.getBoundAddress().boundAddresses();
             assertTrue("gRPC transport should be bound to at least one address", boundAddresses.length > 0);
 
             // Log the bound addresses for debugging
@@ -56,4 +66,15 @@ public void testGrpcTransportStarted() {
             }
         }
     }
+
+    public void testStartGrpcTransportClusterHealth() throws Exception {
+        // REST api cluster health
+        ClusterHealthResponse healthResponse = client().admin().cluster().prepareHealth().get();
+        assertEquals(ClusterHealthStatus.GREEN, healthResponse.getStatus());
+
+        // gRPC transport service health
+        try (NettyGrpcClient client = new NettyGrpcClient.Builder().setAddress(randomNetty4GrpcServerTransportAddr()).build()) {
+            assertEquals(client.checkHealth(), HealthCheckResponse.ServingStatus.SERVING);
+        }
+    }
 }