Experimental gRPC server

buildSrc/src/main/groovy/org/opensearch/gradle/precommit/LicenseHeadersTask.groovy

@@ -147,8 +147,10 @@ class LicenseHeadersTask extends AntTask {
             // Generated resources
             substringMatcher(licenseFamilyCategory: "GEN  ",
                              licenseFamilyName:     "Generated") {
-               // parsers generated by antlr
-               pattern(substring: "ANTLR GENERATED CODE")
+                // parsers generated by antlr
+                pattern(substring: "ANTLR GENERATED CODE")
+                // Protobuf
+                pattern(substring: "Generated by the protocol buffer compiler")
             }
 
             // Vendored Code

gradle/forbidden-dependencies.gradle

@@ -13,7 +13,8 @@
 // we do not want any of these dependencies on the compilation classpath
 // because they could then be used within OpenSearch
 List<String> FORBIDDEN_DEPENDENCIES = [
-  'guava'
+  // TODO: Why is guava forbidden?
+  // 'guava'
 ]
 
 Closure checkDeps = { Configuration configuration ->

gradle/libs.versions.toml

@@ -20,13 +20,20 @@ jettison          = "1.5.4"
 woodstox          = "6.4.0"
 kotlin            = "1.7.10"
 antlr4            = "4.13.1"
-guava             = "32.1.1-jre"
-protobuf          = "3.25.5"
 jakarta_annotation = "1.3.5"
 google_http_client = "1.44.1"
 tdigest           = "3.3"
 hdrhistogram      = "2.2.2"
-grpc              = "1.68.0"
+
+# gRPC
+grpc              = "1.68.1"
+protobuf          = "3.25.5"
+guava             = "32.1.1-jre"
+jsr305            = "3.0.2"
+failureaccess     = "1.0.1"
+error_prone_annotations = "2.24.1"
+javax_annotations = "1.3.2"
+perfmark_api      = "0.26.0"
 
 # when updating the JNA version, also update the version in buildSrc/build.gradle
 jna               = "5.13.0"

modules/transport-netty4/build.gradle

@@ -65,6 +65,22 @@ dependencies {
   api "io.netty:netty-resolver:${versions.netty}"
   api "io.netty:netty-transport:${versions.netty}"
   api "io.netty:netty-transport-native-unix-common:${versions.netty}"
+
+  // gRPC
+  compileOnly "javax.annotation:javax.annotation-api:${versions.javax_annotations}"
+  compileOnly "com.google.code.findbugs:jsr305:${versions.jsr305}"
+  runtimeOnly "com.google.guava:guava:${versions.guava}"
+  api "com.google.guava:failureaccess:${versions.failureaccess}"
+  api "com.google.errorprone:error_prone_annotations:${versions.error_prone_annotations}"
+  api "io.perfmark:perfmark-api:${versions.perfmark_api}"
+  api "io.grpc:grpc-all:${versions.grpc}"
+  api "io.grpc:grpc-netty:${versions.grpc}"
+  api "io.grpc:grpc-api:${versions.grpc}"
+  api "io.grpc:grpc-protobuf-lite:${versions.grpc}"
+  api "io.grpc:grpc-protobuf:${versions.grpc}"
+  api "io.grpc:grpc-stub:${versions.grpc}"
+  api "io.grpc:grpc-core:${versions.grpc}"
+  api "io.grpc:grpc-services:${versions.grpc}"
 }
 
 restResources {

modules/transport-netty4/licenses/error_prone_annotations-2.24.1.jar.sha1

@@ -0,0 +1 @@
+32b299e45105aa9b0df8279c74dc1edfcf313ff0

modules/transport-netty4/licenses/failureaccess-1.0.2.jar.sha1

@@ -0,0 +1 @@
+c4a06a64e650562f30b7bf9aaec1bfed43aca12b

modules/transport-netty4/licenses/grpc-all-1.68.0.jar.sha1

@@ -0,0 +1 @@
+a150a5dac0c120f8da34d59c7730d4a5fe34ec8c

modules/transport-netty4/licenses/grpc-api-1.68.0.jar.sha1

@@ -0,0 +1 @@
+9a9f25c58d8d5b0fcf37ae889a50fec87e34ac08

modules/transport-netty4/licenses/grpc-core-1.68.0.jar.sha1

@@ -0,0 +1 @@
+e5630dfd653d7cad78caf7166e36973f55822e6c

modules/transport-netty4/licenses/grpc-netty-1.68.0.jar.sha1

@@ -0,0 +1 @@
+0ac762f09db8e74f9b17fff5f7f1bcf9a13c5620

modules/transport-netty4/licenses/grpc-protobuf-1.68.0.jar.sha1

@@ -0,0 +1 @@
+b8c1772b35292a853f0707a3512090a8edad1fed

modules/transport-netty4/licenses/grpc-protobuf-lite-1.68.0.jar.sha1

@@ -0,0 +1 @@
+fe6a5349fd76e811c19f16e2b3e9453ee339df4b

modules/transport-netty4/licenses/grpc-services-1.68.0.jar.sha1

@@ -0,0 +1 @@
+b3aa84b0e7cbe4135d27644dd25e3f1cd4eeed85

modules/transport-netty4/licenses/grpc-stub-1.68.0.jar.sha1

@@ -0,0 +1 @@
+ebd89ad550b74724b1bbe8a04203921ebac5e6d4

modules/transport-netty4/licenses/guava-32.1.1-jre.jar.sha1

@@ -0,0 +1 @@
+ad575652d84153075dd41ec6177ccb15251262b2

modules/transport-netty4/licenses/perfmark-api-0.26.0.jar.sha1

@@ -0,0 +1 @@
+ef65452adaf20bf7d12ef55913aba24037b82738

modules/transport-netty4/src/main/java/org/opensearch/grpc/netty4/Netty4GrpcServerTransport.java

@@ -0,0 +1,182 @@
+/*
+ * 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.grpc.netty4;
+
+import io.grpc.BindableService;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+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.core.common.transport.TransportAddress;
+import org.opensearch.grpc.AbstractGrpcServerTransport;
+import org.opensearch.grpc.GrpcStats;
+import org.opensearch.grpc.services.GrpcServiceRegistry;
+import org.opensearch.transport.NettyAllocator;
+import org.opensearch.transport.SharedGroupFactory;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+import io.grpc.ForwardingServerCall;
+import io.grpc.ForwardingServerCallListener;
+import io.grpc.Metadata;
+import io.grpc.Server;
+import io.grpc.ServerCall;
+import io.grpc.ServerCallHandler;
+import io.grpc.ServerInterceptor;
+import io.grpc.Status;
+import io.grpc.netty.NettyServerBuilder;
+import io.grpc.protobuf.services.HealthStatusManager;
+import io.grpc.protobuf.services.ProtoReflectionService;
+
+public class Netty4GrpcServerTransport extends AbstractGrpcServerTransport {
+    private static final Logger logger = LogManager.getLogger(Netty4GrpcServerTransport.class);
+
+    public static final Setting<Integer> SETTING_GRPC_WORKER_COUNT = Setting.intSetting("grpc.worker_count", 1, Setting.Property.NodeScope);
+
+    private final SharedGroupFactory sharedGroupFactory;
+    private final GrpcServiceRegistry grpcServiceRegistry;
+    private final CopyOnWriteArrayList<Server> servers = new CopyOnWriteArrayList<>();
+    private volatile SharedGroupFactory.SharedGroup sharedGroup;
+    private final ServerStatsInterceptor sharedServerStatsInterceptor;
+    private final AtomicLong currentOpen = new AtomicLong(0);
+    private final AtomicLong totalOpened = new AtomicLong(0);
+
+    public Netty4GrpcServerTransport(Settings settings, NetworkService networkService, SharedGroupFactory sharedGroupFactory, GrpcServiceRegistry grpcServiceRegistry) {
+        super(settings, networkService);
+        this.sharedGroupFactory = sharedGroupFactory;
+        this.sharedServerStatsInterceptor = new ServerStatsInterceptor(currentOpen, totalOpened);
+        this.grpcServiceRegistry = grpcServiceRegistry;
+    }
+
+    @Override
+    protected void doStart() {
+        boolean success = false;
+        try {
+            sharedGroup = sharedGroupFactory.getGRPCGroup();
+            bindServer();
+            success = true;
+            logger.info("Started gRPC server on port {}", port);
+        } finally {
+            if (!success) {
+                doStop();
+            }
+        }
+    }
+
+    @Override
+    protected TransportAddress bindAddress(InetAddress hostAddress, PortsRange portRange) {
+        AtomicReference<Exception> lastException = new AtomicReference<>();
+        AtomicReference<TransportAddress> addr = new AtomicReference<>();
+
+        boolean success = portRange.iterate(portNumber -> {
+            try {
+                InetSocketAddress address = new InetSocketAddress(hostAddress, portNumber);
+                NettyServerBuilder srvBuilder = NettyServerBuilder.forAddress(address)
+                    .bossEventLoopGroup(sharedGroup.getLowLevelGroup())
+                    .workerEventLoopGroup(sharedGroup.getLowLevelGroup())
+                    .channelType(NettyAllocator.getServerChannelType())
+                    .intercept(this.sharedServerStatsInterceptor)
+                    .addService(new HealthStatusManager().getHealthService())
+                    .addService(ProtoReflectionService.newInstance());
+
+                for (BindableService bService : grpcServiceRegistry.getServices()) {
+                    srvBuilder.addService(bService);
+                }
+
+                Server srv = srvBuilder.build().start();
+                servers.add(srv);
+                addr.set(new TransportAddress(hostAddress, portNumber));
+                logger.debug("Bound gRPC to address {{}}", address);
+                return true;
+            } catch (Exception e) {
+                lastException.set(e);
+                return false;
+            }
+        });
+
+        if (!success) {
+            throw new RuntimeException("Failed to bind to " + hostAddress + " on ports " + portRange, lastException.get());
+        }
+
+        return addr.get();
+    }
+
+    @Override
+    protected void doStop() {
+        for (Server server : servers) {
+            if (server != null) {
+                server.shutdown();
+                try {
+                    server.awaitTermination(30, TimeUnit.SECONDS);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    logger.warn("Interrupted while shutting down gRPC server");
+                } finally {
+                    server.shutdownNow();
+                }
+            }
+
+            if (sharedGroup != null) {
+                sharedGroup.shutdown();
+                sharedGroup = null;
+            }
+        }
+    }
+
+    @Override
+    protected void doClose() {}
+
+    @Override
+    public GrpcStats stats() {
+        return new GrpcStats(totalOpened.get(), currentOpen.get());
+    }
+
+    static class ServerStatsInterceptor implements ServerInterceptor {
+        private final AtomicLong currentOpen;
+        private final AtomicLong totalOpened;
+
+        ServerStatsInterceptor(AtomicLong currentOpen, AtomicLong totalOpened) {
+            this.currentOpen = currentOpen;
+            this.totalOpened = totalOpened;
+        }
+
+        @Override
+        public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
+            ServerCall<ReqT, RespT> call,
+            Metadata headers,
+            ServerCallHandler<ReqT, RespT> next
+        ) {
+            logger.debug("Intercepted call - Method: {}, Authority: {}, Headers: {}",
+                call.getMethodDescriptor().getFullMethodName(),
+                call.getAuthority(),
+                headers);
+
+            currentOpen.incrementAndGet();
+            totalOpened.incrementAndGet();
+
+            return new ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT>(
+                next.startCall(new ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT>(call) {
+                    @Override
+                    public void close(Status status, Metadata trailers) {
+                        currentOpen.decrementAndGet();
+                        super.close(status, trailers);
+                    }
+                }, headers)
+            ) {
+            };
+        }
+    }
+}

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

@@ -44,6 +44,9 @@
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.core.indices.breaker.CircuitBreakerService;
 import org.opensearch.core.xcontent.NamedXContentRegistry;
+import org.opensearch.grpc.GrpcServerTransport;
+import org.opensearch.grpc.netty4.Netty4GrpcServerTransport;
+import org.opensearch.grpc.services.GrpcServiceRegistry;
 import org.opensearch.http.HttpServerTransport;
 import org.opensearch.http.netty4.Netty4HttpServerTransport;
 import org.opensearch.http.netty4.ssl.SecureNetty4HttpServerTransport;
@@ -68,6 +71,7 @@ public class Netty4ModulePlugin extends Plugin implements NetworkPlugin {
     public static final String NETTY_SECURE_TRANSPORT_NAME = "netty4-secure";
     public static final String NETTY_HTTP_TRANSPORT_NAME = "netty4";
     public static final String NETTY_SECURE_HTTP_TRANSPORT_NAME = "netty4-secure";
+    public static final String NETTY_GRPC_TRANSPORT_NAME = "netty4-grpc";
 
     private final SetOnce<SharedGroupFactory> groupFactory = new SetOnce<>();
 
@@ -150,6 +154,14 @@ public Map<String, Supplier<HttpServerTransport>> getHttpTransports(
         );
     }
 
+    @Override
+    public Map<String, Supplier<GrpcServerTransport>> getGrpcTransports(Settings settings, NetworkService networkService, GrpcServiceRegistry grpcServiceRegistry) {
+        return Collections.singletonMap(
+            NETTY_GRPC_TRANSPORT_NAME,
+            () -> new Netty4GrpcServerTransport(settings, networkService, getSharedGroupFactory(settings), grpcServiceRegistry)
+        );
+    }
+
     @Override
     public Map<String, Supplier<HttpServerTransport>> getSecureHttpTransports(
         Settings settings,

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

@@ -36,6 +36,7 @@
 import org.apache.logging.log4j.Logger;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.util.concurrent.AbstractRefCounted;
+import org.opensearch.grpc.netty4.Netty4GrpcServerTransport;
 import org.opensearch.http.HttpServerTransport;
 import org.opensearch.http.netty4.Netty4HttpServerTransport;
 import org.opensearch.transport.netty4.Netty4Transport;
@@ -62,14 +63,17 @@ public final class SharedGroupFactory {
     private final Settings settings;
     private final int workerCount;
     private final int httpWorkerCount;
+    private final int grpcWorkerCount;
 
     private RefCountedGroup genericGroup;
     private SharedGroup dedicatedHttpGroup;
+    private SharedGroup dedicatedGRPCGroup;
 
     public SharedGroupFactory(Settings settings) {
         this.settings = settings;
         this.workerCount = Netty4Transport.WORKER_COUNT.get(settings);
         this.httpWorkerCount = Netty4HttpServerTransport.SETTING_HTTP_WORKER_COUNT.get(settings);
+        this.grpcWorkerCount = Netty4GrpcServerTransport.SETTING_GRPC_WORKER_COUNT.get(settings);
     }
 
     public Settings getSettings() {
@@ -99,6 +103,21 @@ public synchronized SharedGroup getHttpGroup() {
         }
     }
 
+    public synchronized SharedGroup getGRPCGroup() {
+        if (grpcWorkerCount == 0) {
+            return getGenericGroup();
+        } else {
+            if (dedicatedGRPCGroup == null) {
+                NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(
+                    grpcWorkerCount,
+                    daemonThreadFactory(settings, HttpServerTransport.HTTP_SERVER_WORKER_THREAD_NAME_PREFIX)
+                );
+                dedicatedGRPCGroup = new SharedGroup(new RefCountedGroup(eventLoopGroup));
+            }
+            return dedicatedGRPCGroup;
+        }
+    }
+
     private SharedGroup getGenericGroup() {
         if (genericGroup == null) {
             EventLoopGroup eventLoopGroup = new NioEventLoopGroup(