diff --git a/bom/application/pom.xml b/bom/application/pom.xml index 05c587b5388c95..5fceb8ee039018 100644 --- a/bom/application/pom.xml +++ b/bom/application/pom.xml @@ -1906,6 +1906,11 @@ + + io.quarkus + quarkus-grpc-xds + ${project.version} + io.quarkus quarkus-grpc-api diff --git a/extensions/grpc/pom.xml b/extensions/grpc/pom.xml index 27cb1447a3e968..1d41c8feb45663 100644 --- a/extensions/grpc/pom.xml +++ b/extensions/grpc/pom.xml @@ -20,5 +20,6 @@ stubs deployment runtime + xds \ No newline at end of file diff --git a/extensions/grpc/runtime/pom.xml b/extensions/grpc/runtime/pom.xml index 6eef11e3bd698b..9a0404be396053 100644 --- a/extensions/grpc/runtime/pom.xml +++ b/extensions/grpc/runtime/pom.xml @@ -79,36 +79,6 @@ - - io.grpc - grpc-xds - - - commons-logging - commons-logging - - - com.google.auto.value - auto-value-annotations - - - com.google.android - annotations - - - com.google.code.findbugs - jsr305 - - - org.codehaus.mojo - animal-sniffer-annotations - - - org.checkerframework - checker-qual - - - io.quarkus quarkus-mutiny diff --git a/extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/GrpcServerRecorder.java b/extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/GrpcServerRecorder.java index 2c759668dd48c3..9291396794922d 100644 --- a/extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/GrpcServerRecorder.java +++ b/extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/GrpcServerRecorder.java @@ -18,7 +18,6 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; -import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; @@ -31,25 +30,19 @@ import grpc.health.v1.HealthOuterClass; import io.grpc.BindableService; -import io.grpc.InsecureServerCredentials; -import io.grpc.Metadata; import io.grpc.Server; import io.grpc.ServerBuilder; -import io.grpc.ServerCall; -import io.grpc.ServerCallHandler; -import io.grpc.ServerCredentials; import io.grpc.ServerInterceptor; import io.grpc.ServerInterceptors; import io.grpc.ServerMethodDefinition; import io.grpc.ServerServiceDefinition; -import io.grpc.xds.XdsServerBuilder; -import io.grpc.xds.XdsServerCredentials; import io.quarkus.arc.Arc; import io.quarkus.arc.Subclass; import io.quarkus.grpc.runtime.config.GrpcConfiguration; import io.quarkus.grpc.runtime.config.GrpcServerConfiguration; import io.quarkus.grpc.runtime.config.GrpcServerNettyConfig; import io.quarkus.grpc.runtime.config.Xds; +import io.quarkus.grpc.runtime.devmode.DevModeInterceptor; import io.quarkus.grpc.runtime.devmode.GrpcHotReplacementInterceptor; import io.quarkus.grpc.runtime.devmode.GrpcServerReloader; import io.quarkus.grpc.runtime.devmode.XdsServerReloader; @@ -57,6 +50,7 @@ import io.quarkus.grpc.runtime.reflection.ReflectionService; import io.quarkus.grpc.runtime.supports.CompressionInterceptor; import io.quarkus.grpc.runtime.supports.blocking.BlockingServerInterceptor; +import io.quarkus.grpc.spi.GrpcBuilderProvider; import io.quarkus.runtime.LaunchMode; import io.quarkus.runtime.QuarkusBindException; import io.quarkus.runtime.RuntimeValue; @@ -68,8 +62,6 @@ import io.vertx.core.Handler; import io.vertx.core.Promise; import io.vertx.core.Vertx; -import io.vertx.core.impl.EventLoopContext; -import io.vertx.core.impl.VertxInternal; import io.vertx.ext.web.Route; import io.vertx.ext.web.Router; import io.vertx.ext.web.RoutingContext; @@ -428,34 +420,15 @@ public RuntimeValue initServerInterceptorStorage( private Map.Entry buildServer(Vertx vertx, GrpcServerConfiguration configuration, Map> blockingMethodsPerService, GrpcContainer grpcContainer, LaunchMode launchMode) { - Xds xds = configuration.xds; - boolean isXDS = xds != null && xds.enabled; - int port = launchMode == LaunchMode.TEST ? configuration.testPort : configuration.port; AtomicBoolean usePlainText = new AtomicBoolean(); + GrpcBuilderProvider provider = GrpcBuilderProvider.findServerBuilderProvider(configuration); + ServerBuilder builder; - if (isXDS) { - ServerCredentials credentials = InsecureServerCredentials.create(); - if (xds.secure) { - credentials = XdsServerCredentials.create(credentials); - } - builder = XdsServerBuilder.forPort(port, credentials); - // wrap with Vert.x context, so that the context interceptors work - VertxInternal vxi = (VertxInternal) vertx; - Executor delegate = vertx.nettyEventLoopGroup(); - EventLoopContext context = vxi.createEventLoopContext(); - Executor executor = command -> delegate.execute(() -> { - context.dispatch(command); - }); - builder.executor(executor); - // custom XDS interceptors - if (launchMode == LaunchMode.DEVELOPMENT) { - builder.intercept(new DevModeInterceptor(Thread.currentThread().getContextClassLoader())); - // TODO -- see GrpcHotReplacementInterceptor usage -- OK? - builder.intercept(new GrpcHotReplacementInterceptor()); - } + if (provider != null) { + builder = provider.createServerBuilder(vertx, configuration, launchMode); } else { VertxServerBuilder vsBuilder = VertxServerBuilder.forAddress(vertx, configuration.host, port); // add Vert.x specific stuff here @@ -514,7 +487,7 @@ private Map.Entry buildServer(Vertx vertx, GrpcServerConfigurat LOGGER.debugf("Starting gRPC Server on %s:%d [%s] ...", configuration.host, port, - isXDS ? "XDS enabled" : "SSL enabled: " + !usePlainText.get()); + provider != null ? provider.serverInfo() : "SSL enabled: " + !usePlainText.get()); return new AbstractMap.SimpleEntry<>(port, builder.build()); } @@ -658,24 +631,4 @@ public void run(Runnable command) { } } } - - private static class DevModeInterceptor implements ServerInterceptor { - private final ClassLoader classLoader; - - public DevModeInterceptor(ClassLoader contextClassLoader) { - classLoader = contextClassLoader; - } - - @Override - public ServerCall.Listener interceptCall(ServerCall serverCall, Metadata metadata, - ServerCallHandler next) { - ClassLoader originalTccl = Thread.currentThread().getContextClassLoader(); - Thread.currentThread().setContextClassLoader(classLoader); - try { - return next.startCall(serverCall, metadata); - } finally { - Thread.currentThread().setContextClassLoader(originalTccl); - } - } - } } diff --git a/extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/devmode/DevModeInterceptor.java b/extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/devmode/DevModeInterceptor.java new file mode 100644 index 00000000000000..9ecf69d9967a4a --- /dev/null +++ b/extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/devmode/DevModeInterceptor.java @@ -0,0 +1,26 @@ +package io.quarkus.grpc.runtime.devmode; + +import io.grpc.Metadata; +import io.grpc.ServerCall; +import io.grpc.ServerCallHandler; +import io.grpc.ServerInterceptor; + +public class DevModeInterceptor implements ServerInterceptor { + private final ClassLoader classLoader; + + public DevModeInterceptor(ClassLoader contextClassLoader) { + classLoader = contextClassLoader; + } + + @Override + public ServerCall.Listener interceptCall(ServerCall serverCall, Metadata metadata, + ServerCallHandler next) { + ClassLoader originalTccl = Thread.currentThread().getContextClassLoader(); + Thread.currentThread().setContextClassLoader(classLoader); + try { + return next.startCall(serverCall, metadata); + } finally { + Thread.currentThread().setContextClassLoader(originalTccl); + } + } +} diff --git a/extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/supports/Channels.java b/extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/supports/Channels.java index 994b9573e249a0..947e0a1ef1f412 100644 --- a/extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/supports/Channels.java +++ b/extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/supports/Channels.java @@ -4,7 +4,6 @@ import static io.grpc.internal.GrpcUtil.DEFAULT_MAX_MESSAGE_SIZE; import static io.grpc.netty.NettyChannelBuilder.DEFAULT_FLOW_CONTROL_WINDOW; import static io.quarkus.grpc.runtime.config.GrpcClientConfiguration.DNS; -import static io.quarkus.grpc.runtime.config.GrpcClientConfiguration.XDS; import java.io.IOException; import java.io.InputStream; @@ -31,19 +30,15 @@ import io.grpc.CallOptions; import io.grpc.Channel; -import io.grpc.ChannelCredentials; import io.grpc.ClientCall; import io.grpc.ClientInterceptor; import io.grpc.ClientInterceptors; -import io.grpc.Grpc; -import io.grpc.InsecureChannelCredentials; import io.grpc.ManagedChannel; import io.grpc.ManagedChannelBuilder; import io.grpc.MethodDescriptor; import io.grpc.netty.GrpcSslContexts; import io.grpc.netty.NegotiationType; import io.grpc.netty.NettyChannelBuilder; -import io.grpc.xds.XdsChannelCredentials; import io.netty.handler.ssl.SslContext; import io.netty.handler.ssl.SslContextBuilder; import io.quarkus.arc.Arc; @@ -56,8 +51,8 @@ import io.quarkus.grpc.runtime.config.GrpcClientConfiguration; import io.quarkus.grpc.runtime.config.GrpcServerConfiguration; import io.quarkus.grpc.runtime.config.SslClientConfig; -import io.quarkus.grpc.runtime.config.Xds; import io.quarkus.grpc.runtime.stork.StorkMeasuringGrpcInterceptor; +import io.quarkus.grpc.spi.GrpcBuilderProvider; import io.quarkus.runtime.LaunchMode; import io.quarkus.runtime.util.ClassPathUtils; import io.smallrye.mutiny.infrastructure.Infrastructure; @@ -101,21 +96,23 @@ public static Channel createChannel(String name, Set perClientIntercepto throw new IllegalStateException("gRPC client " + name + " is missing configuration."); } + GrpcBuilderProvider provider = GrpcBuilderProvider.findChannelBuilderProvider(config); + boolean vertxGrpc = config.useQuarkusGrpcClient; - Xds xdsConfig = config.xds; String host = config.host; int port = config.port; String nameResolver = config.nameResolver; boolean stork = Stork.STORK.equalsIgnoreCase(nameResolver); - boolean xds = XDS.equalsIgnoreCase(nameResolver) || (xdsConfig != null && xdsConfig.enabled); String[] resolverSplit = nameResolver.split(":"); - String resolver = xds ? XDS : resolverSplit[0]; + String resolver = provider != null ? provider.resolver() : resolverSplit[0]; // TODO -- does this work for Vert.x gRPC client? - if (!vertxGrpc && (DNS.equalsIgnoreCase(resolver) || xds)) { + if (provider != null) { + host = provider.adjustHost(host); + } else if (!vertxGrpc && DNS.equalsIgnoreCase(resolver)) { host = "/" + host; // dns or xds name resolver needs triple slash at the beginning } @@ -137,16 +134,8 @@ public static Channel createChannel(String name, Set perClientIntercepto String target = String.format("%s://%s:%d", resolver, host, port); LOGGER.debugf("Target for client '%s': %s", name, target); - boolean xdsSecure = false; - if (xds) { - if (xdsConfig != null) { - target = xdsConfig.target.orElse(target); - xdsSecure = xdsConfig.secure; - } - } - SslContext context = null; - if (!plainText && !xds) { + if (!plainText && provider == null) { Path trustStorePath = config.ssl.trustStore.orElse(null); Path certificatePath = config.ssl.certificate.orElse(null); Path keyPath = config.ssl.key.orElse(null); @@ -173,13 +162,9 @@ public static Channel createChannel(String name, Set perClientIntercepto String loadBalancingPolicy = stork ? Stork.STORK : config.loadBalancingPolicy; - ManagedChannelBuilder builder; - if (xds) { - ChannelCredentials credentials = InsecureChannelCredentials.create(); - if (xdsSecure) { - credentials = XdsChannelCredentials.create(credentials); - } - builder = Grpc.newChannelBuilder(target, credentials); + ManagedChannelBuilder builder; + if (provider != null) { + builder = provider.createChannelBuilder(config, target); } else { builder = NettyChannelBuilder .forTarget(target) @@ -233,7 +218,7 @@ public static Channel createChannel(String name, Set perClientIntercepto builder.keepAliveTimeout(idleTimeout.get().toMillis(), TimeUnit.MILLISECONDS); } - if (plainText && !xds) { + if (plainText && provider == null) { builder.usePlaintext(); } if (context != null && (builder instanceof NettyChannelBuilder)) { @@ -244,7 +229,8 @@ public static Channel createChannel(String name, Set perClientIntercepto interceptorContainer.getSortedPerServiceInterceptors(perClientInterceptors).forEach(builder::intercept); interceptorContainer.getSortedGlobalInterceptors().forEach(builder::intercept); - LOGGER.info(String.format("Creating %s gRPC channel ...", xds ? "XDS" : "Netty")); + LOGGER.info(String.format("Creating %s gRPC channel ...", + provider != null ? provider.channelInfo() : "Netty")); return builder.build(); } else { diff --git a/extensions/grpc/runtime/src/main/java/io/quarkus/grpc/spi/GrpcBuilderProvider.java b/extensions/grpc/runtime/src/main/java/io/quarkus/grpc/spi/GrpcBuilderProvider.java new file mode 100644 index 00000000000000..08418262e9a6a1 --- /dev/null +++ b/extensions/grpc/runtime/src/main/java/io/quarkus/grpc/spi/GrpcBuilderProvider.java @@ -0,0 +1,134 @@ +package io.quarkus.grpc.spi; + +import java.util.ServiceLoader; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.grpc.ManagedChannelBuilder; +import io.grpc.ServerBuilder; +import io.quarkus.grpc.runtime.config.GrpcClientConfiguration; +import io.quarkus.grpc.runtime.config.GrpcServerConfiguration; +import io.quarkus.runtime.LaunchMode; +import io.vertx.core.Vertx; + +/** + * Allow for additional types of gRPC server and channels to be used / built. + * + * This is an experimental SPI, subject to change. + */ +public interface GrpcBuilderProvider> { + + Logger log = LoggerFactory.getLogger(GrpcBuilderProvider.class); + + /** + * Find gRPC server builder provider. + * + * @param configuration the gRPC server configuration + * @return provider instance or null if none is provided + */ + @SuppressWarnings("rawtypes") + static GrpcBuilderProvider findServerBuilderProvider(GrpcServerConfiguration configuration) { + GrpcBuilderProvider provider = null; + ServiceLoader providers = ServiceLoader.load(GrpcBuilderProvider.class); + for (GrpcBuilderProvider p : providers) { + if (p.providesServer(configuration)) { + if (provider != null) { + throw new IllegalArgumentException("Too many GrpcBuilderProviders enabled: " + providers); + } + log.info("Found server GrpcBuilderProvider: {}", p); + provider = p; + } + } + return provider; + } + + /** + * Find gRPC client builder provider. + * + * @param configuration the gRPC client configuration + * @return provider instance or null if none is provided + */ + @SuppressWarnings("rawtypes") + static GrpcBuilderProvider findChannelBuilderProvider(GrpcClientConfiguration configuration) { + GrpcBuilderProvider provider = null; + ServiceLoader providers = ServiceLoader.load(GrpcBuilderProvider.class); + for (GrpcBuilderProvider p : providers) { + if (p.providesChannel(configuration)) { + if (provider != null) { + throw new IllegalArgumentException("Too many GrpcBuilderProviders enabled: " + providers); + } + log.info("Found channel GrpcBuilderProvider: {}", p); + provider = p; + } + } + return provider; + } + + /** + * Does this builder provider provide a new gRPC server instance. + * + * @param configuration the gRPC server configuration + * @return true if yes, false if no + */ + boolean providesServer(GrpcServerConfiguration configuration); + + /** + * Create initial server builder. + * + * @param vertx the Vertx instance + * @param configuration the gRPC server configuration + * @param launchMode current launch mode + * @return new ServerBuilder instance + */ + ServerBuilder createServerBuilder(Vertx vertx, GrpcServerConfiguration configuration, LaunchMode launchMode); + + /** + * Provide server info. + * + * @return simple server info + */ + String serverInfo(); + + /** + * Does this builder provider provide a new gRPC channel instance. + * + * @param configuration the gRPC client configuration + * @return true if yes, false if no + */ + boolean providesChannel(GrpcClientConfiguration configuration); + + /** + * Get resolver. + * + * @return the resolver + */ + String resolver(); + + /** + * Adjust host, if needed. + * By default, no adjustment is made. + * + * @param host the host + * @return adjusted host, if needed + */ + default String adjustHost(String host) { + return host; + } + + /** + * Create initial channel builder. + * + * @param configuration the gRPC client configuration + * @param target the channel target + * @return new ChannelBuilder + */ + ManagedChannelBuilder createChannelBuilder(GrpcClientConfiguration configuration, String target); + + /** + * Provide channel info. + * + * @return simple channel info + */ + String channelInfo(); +} diff --git a/extensions/grpc/xds/pom.xml b/extensions/grpc/xds/pom.xml new file mode 100644 index 00000000000000..ae52996721126f --- /dev/null +++ b/extensions/grpc/xds/pom.xml @@ -0,0 +1,51 @@ + + + 4.0.0 + + quarkus-grpc-parent + io.quarkus + 999-SNAPSHOT + + + quarkus-grpc-xds + Quarkus - gRPC - xDS + gRPC xDS support + + + io.quarkus + quarkus-grpc + + + io.grpc + grpc-xds + + + commons-logging + commons-logging + + + com.google.auto.value + auto-value-annotations + + + com.google.android + annotations + + + com.google.code.findbugs + jsr305 + + + org.codehaus.mojo + animal-sniffer-annotations + + + org.checkerframework + checker-qual + + + + + diff --git a/extensions/grpc/xds/src/main/java/io/quarkus/grpc/xds/XdsGrpcServerBuilderProvider.java b/extensions/grpc/xds/src/main/java/io/quarkus/grpc/xds/XdsGrpcServerBuilderProvider.java new file mode 100644 index 00000000000000..81c463c857a48d --- /dev/null +++ b/extensions/grpc/xds/src/main/java/io/quarkus/grpc/xds/XdsGrpcServerBuilderProvider.java @@ -0,0 +1,98 @@ +package io.quarkus.grpc.xds; + +import static io.quarkus.grpc.runtime.config.GrpcClientConfiguration.XDS; + +import java.util.concurrent.Executor; + +import io.grpc.ChannelCredentials; +import io.grpc.Grpc; +import io.grpc.InsecureChannelCredentials; +import io.grpc.InsecureServerCredentials; +import io.grpc.ManagedChannelBuilder; +import io.grpc.ServerBuilder; +import io.grpc.ServerCredentials; +import io.grpc.xds.XdsChannelCredentials; +import io.grpc.xds.XdsServerBuilder; +import io.grpc.xds.XdsServerCredentials; +import io.quarkus.grpc.runtime.config.GrpcClientConfiguration; +import io.quarkus.grpc.runtime.config.GrpcServerConfiguration; +import io.quarkus.grpc.runtime.config.Xds; +import io.quarkus.grpc.runtime.devmode.DevModeInterceptor; +import io.quarkus.grpc.runtime.devmode.GrpcHotReplacementInterceptor; +import io.quarkus.grpc.spi.GrpcBuilderProvider; +import io.quarkus.runtime.LaunchMode; +import io.vertx.core.Vertx; +import io.vertx.core.impl.EventLoopContext; +import io.vertx.core.impl.VertxInternal; + +public class XdsGrpcServerBuilderProvider implements GrpcBuilderProvider { + @Override + public boolean providesServer(GrpcServerConfiguration configuration) { + Xds xds = configuration.xds; + return xds != null && xds.enabled; + } + + @Override + public ServerBuilder createServerBuilder(Vertx vertx, GrpcServerConfiguration configuration, + LaunchMode launchMode) { + Xds xds = configuration.xds; + int port = launchMode == LaunchMode.TEST ? configuration.testPort : configuration.port; + ServerCredentials credentials = InsecureServerCredentials.create(); + if (xds.secure) { + credentials = XdsServerCredentials.create(credentials); + } + ServerBuilder builder = XdsServerBuilder.forPort(port, credentials); + // wrap with Vert.x context, so that the context interceptors work + VertxInternal vxi = (VertxInternal) vertx; + Executor delegate = vertx.nettyEventLoopGroup(); + EventLoopContext context = vxi.createEventLoopContext(); + Executor executor = command -> delegate.execute(() -> context.dispatch(command)); + builder.executor(executor); + // custom XDS interceptors + if (launchMode == LaunchMode.DEVELOPMENT) { + builder.intercept(new DevModeInterceptor(Thread.currentThread().getContextClassLoader())); + builder.intercept(new GrpcHotReplacementInterceptor()); + } + return builder; + } + + @Override + public String serverInfo() { + return "xDS enabled"; + } + + @Override + public boolean providesChannel(GrpcClientConfiguration configuration) { + Xds xds = configuration.xds; + if (xds != null) { + return xds.enabled || XDS.equalsIgnoreCase(configuration.nameResolver); + } else { + return false; + } + } + + @Override + public String resolver() { + return XDS; + } + + @Override + public String adjustHost(String host) { + return "/" + host; + } + + @Override + public ManagedChannelBuilder createChannelBuilder(GrpcClientConfiguration configuration, String target) { + Xds xds = configuration.xds; + ChannelCredentials credentials = InsecureChannelCredentials.create(); + if (xds.secure) { + credentials = XdsChannelCredentials.create(credentials); + } + return Grpc.newChannelBuilder(target, credentials); + } + + @Override + public String channelInfo() { + return "xDS"; + } +} diff --git a/extensions/grpc/xds/src/main/resources/META-INF/services/io.quarkus.grpc.spi.GrpcBuilderProvider b/extensions/grpc/xds/src/main/resources/META-INF/services/io.quarkus.grpc.spi.GrpcBuilderProvider new file mode 100644 index 00000000000000..d54d69a8d823e9 --- /dev/null +++ b/extensions/grpc/xds/src/main/resources/META-INF/services/io.quarkus.grpc.spi.GrpcBuilderProvider @@ -0,0 +1 @@ +io.quarkus.grpc.xds.XdsGrpcServerBuilderProvider \ No newline at end of file diff --git a/integration-tests/istio/maven-invoker-way/src/it/xds-grpc/pom.xml b/integration-tests/istio/maven-invoker-way/src/it/xds-grpc/pom.xml index 615a2c8ff252f2..844e6b1e012003 100644 --- a/integration-tests/istio/maven-invoker-way/src/it/xds-grpc/pom.xml +++ b/integration-tests/istio/maven-invoker-way/src/it/xds-grpc/pom.xml @@ -26,7 +26,7 @@ io.quarkus - quarkus-grpc + quarkus-grpc-xds io.quarkus