diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java b/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java index c34d947ab2..9d5a8766f1 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java @@ -21,6 +21,8 @@ import ch.qos.logback.classic.LoggerContext; import ch.qos.logback.classic.joran.JoranConfigurator; import ch.qos.logback.core.joran.spi.JoranException; import java.util.Date; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; import org.apache.rocketmq.broker.BrokerController; import org.apache.rocketmq.broker.BrokerStartup; import org.apache.rocketmq.client.log.ClientLogger; @@ -33,6 +35,8 @@ import org.apache.rocketmq.proxy.common.StartAndShutdown; import org.apache.rocketmq.proxy.config.ConfigurationManager; import org.apache.rocketmq.proxy.config.ProxyConfig; import org.apache.rocketmq.proxy.grpc.GrpcServer; +import org.apache.rocketmq.proxy.grpc.GrpcServerBuilder; +import org.apache.rocketmq.proxy.grpc.v2.GrpcMessagingProcessor; import org.apache.rocketmq.proxy.grpc.v2.adapter.ProxyMode; import org.apache.rocketmq.proxy.grpc.v2.service.ClusterGrpcService; import org.apache.rocketmq.proxy.grpc.v2.service.GrpcForwardService; @@ -59,8 +63,12 @@ public class ProxyStartup { // init thread pool monitor for proxy. initThreadPoolMonitor(); + ThreadPoolExecutor executor = createServerExecutor(); + // create grpcServer - GrpcServer grpcServer = createGrpcServer(); + GrpcServer grpcServer = GrpcServerBuilder.newBuilder(executor) + .addService(createServiceProcessor()) + .build(); PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(grpcServer); // start servers one by one. @@ -85,7 +93,7 @@ public class ProxyStartup { log.info(new Date() + " rmq-proxy startup successfully"); } - private static GrpcServer createGrpcServer() throws Exception { + private static GrpcMessagingProcessor createServiceProcessor() { GrpcForwardService grpcService; String proxyModeStr = ConfigurationManager.getProxyConfig().getProxyMode(); if (ProxyMode.isClusterMode(proxyModeStr)) { @@ -109,7 +117,8 @@ public class ProxyStartup { throw new IllegalArgumentException("try to start grpc server with wrong mode, use 'local' or 'cluster'"); } - return new GrpcServer(grpcService); + PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(grpcService); + return new GrpcMessagingProcessor(grpcService); } private static BrokerController createBrokerController() { @@ -117,6 +126,20 @@ public class ProxyStartup { return BrokerStartup.createBrokerController(brokerStartupArgs); } + public static ThreadPoolExecutor createServerExecutor() { + int threadPoolNums = ConfigurationManager.getProxyConfig().getGrpcThreadPoolNums(); + int threadPoolQueueCapacity = ConfigurationManager.getProxyConfig().getGrpcThreadPoolQueueCapacity(); + ThreadPoolExecutor executor = ThreadPoolMonitor.createAndMonitor( + threadPoolNums, + threadPoolNums, + 1, TimeUnit.MINUTES, + "GrpcRequestExecutorThread", + threadPoolQueueCapacity + ); + PROXY_START_AND_SHUTDOWN.appendShutdown(executor::shutdown); + return executor; + } + public static void initThreadPoolMonitor() { ThreadPoolMonitor.init(); ProxyConfig config = ConfigurationManager.getProxyConfig(); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/common/AbstractStartAndShutdown.java b/proxy/src/main/java/org/apache/rocketmq/proxy/common/AbstractStartAndShutdown.java index 35e361fae1..c59f18c4cf 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/common/AbstractStartAndShutdown.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/common/AbstractStartAndShutdown.java @@ -41,4 +41,32 @@ public abstract class AbstractStartAndShutdown implements StartAndShutdown { startAndShutdownList.get(index).shutdown(); } } + + public void appendStart(Start start) { + this.appendStartAndShutdown(new StartAndShutdown() { + @Override + public void shutdown() throws Exception { + + } + + @Override + public void start() throws Exception { + start.start(); + } + }); + } + + public void appendShutdown(Shutdown shutdown) { + this.appendStartAndShutdown(new StartAndShutdown() { + @Override + public void shutdown() throws Exception { + shutdown.shutdown(); + } + + @Override + public void start() throws Exception { + + } + }); + } } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/common/Shutdown.java b/proxy/src/main/java/org/apache/rocketmq/proxy/common/Shutdown.java new file mode 100644 index 0000000000..28f4f92f54 --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/common/Shutdown.java @@ -0,0 +1,22 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.proxy.common; + +public interface Shutdown { + void shutdown() throws Exception; +} diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/common/Start.java b/proxy/src/main/java/org/apache/rocketmq/proxy/common/Start.java new file mode 100644 index 0000000000..3cf74d47d2 --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/common/Start.java @@ -0,0 +1,22 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.proxy.common; + +public interface Start { + void start() throws Exception; +} diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/common/StartAndShutdown.java b/proxy/src/main/java/org/apache/rocketmq/proxy/common/StartAndShutdown.java index ffb2523e97..565e92c25c 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/common/StartAndShutdown.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/common/StartAndShutdown.java @@ -17,7 +17,5 @@ package org.apache.rocketmq.proxy.common; -public interface StartAndShutdown { - void start() throws Exception; - void shutdown() throws Exception; +public interface StartAndShutdown extends Start, Shutdown { } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServer.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServer.java index 78e3770659..d663a88f6d 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServer.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServer.java @@ -17,152 +17,22 @@ package org.apache.rocketmq.proxy.grpc; -import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts; -import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder; -import io.grpc.netty.shaded.io.netty.channel.epoll.EpollEventLoopGroup; -import io.grpc.netty.shaded.io.netty.channel.epoll.EpollServerSocketChannel; -import io.grpc.netty.shaded.io.netty.channel.nio.NioEventLoopGroup; -import io.grpc.netty.shaded.io.netty.channel.socket.nio.NioServerSocketChannel; -import io.grpc.netty.shaded.io.netty.handler.ssl.ClientAuth; -import io.grpc.netty.shaded.io.netty.handler.ssl.util.InsecureTrustManagerFactory; -import io.grpc.netty.shaded.io.netty.handler.ssl.util.SelfSignedCertificate; -import java.io.FileInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.security.cert.CertificateException; -import java.util.List; -import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; -import javax.net.ssl.SSLException; -import org.apache.rocketmq.acl.AccessValidator; import org.apache.rocketmq.common.constant.LoggerName; -import org.apache.rocketmq.common.thread.ThreadPoolMonitor; -import org.apache.rocketmq.common.utils.ServiceProvider; import org.apache.rocketmq.logging.InternalLogger; import org.apache.rocketmq.logging.InternalLoggerFactory; import org.apache.rocketmq.proxy.common.StartAndShutdown; -import org.apache.rocketmq.proxy.config.ConfigurationManager; -import org.apache.rocketmq.proxy.config.ProxyConfig; -import org.apache.rocketmq.proxy.grpc.interceptor.AuthenticationInterceptor; -import org.apache.rocketmq.proxy.grpc.interceptor.ContextInterceptor; -import org.apache.rocketmq.proxy.grpc.interceptor.HeaderInterceptor; -import org.apache.rocketmq.proxy.grpc.v2.GrpcMessagingProcessor; -import org.apache.rocketmq.proxy.grpc.v2.service.GrpcForwardService; public class GrpcServer implements StartAndShutdown { private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); private final io.grpc.Server server; - private final ThreadPoolExecutor executor; - private final GrpcForwardService grpcForwardService; - public GrpcServer(GrpcForwardService grpcForwardService) { - this.grpcForwardService = grpcForwardService; - int port = ConfigurationManager.getProxyConfig().getGrpcServerPort(); - NettyServerBuilder serverBuilder = NettyServerBuilder.forPort(port); - - try { - configSslContext(serverBuilder); - } catch (Exception e) { - log.error("grpc tls set failed. msg: {}, e:", e.getMessage(), e); - throw new RuntimeException("grpc tls set failed: " + e.getMessage()); - } - - // create executor - int threadPoolNums = ConfigurationManager.getProxyConfig().getGrpcThreadPoolNums(); - int threadPoolQueueCapacity = ConfigurationManager.getProxyConfig().getGrpcThreadPoolQueueCapacity(); - this.executor = ThreadPoolMonitor.createAndMonitor( - threadPoolNums, - threadPoolNums, - 1, TimeUnit.MINUTES, - "GrpcRequestExecutorThread", - threadPoolQueueCapacity - ); - - GrpcMessagingProcessor messagingProcessor = createProcessor(); - - // build server - int bossLoopNum = ConfigurationManager.getProxyConfig().getGrpcBossLoopNum(); - int workerLoopNum = ConfigurationManager.getProxyConfig().getGrpcWorkerLoopNum(); - int maxInboundMessageSize = ConfigurationManager.getProxyConfig().getGrpcMaxInboundMessageSize(); - - if (ConfigurationManager.getProxyConfig().isEnableGrpcEpoll()) { - serverBuilder.maxInboundMessageSize(maxInboundMessageSize) - .bossEventLoopGroup(new EpollEventLoopGroup(bossLoopNum)) - .workerEventLoopGroup(new EpollEventLoopGroup(workerLoopNum)) - .channelType(EpollServerSocketChannel.class) - .addService(messagingProcessor) - .executor(this.executor); - } else { - serverBuilder.maxInboundMessageSize(maxInboundMessageSize) - .bossEventLoopGroup(new NioEventLoopGroup(bossLoopNum)) - .workerEventLoopGroup(new NioEventLoopGroup(workerLoopNum)) - .channelType(NioServerSocketChannel.class) - .addService(messagingProcessor) - .executor(this.executor); - } - - configInterceptor(serverBuilder); - this.server = serverBuilder.build(); - - log.info( - "grpc server has built. port: {}, tlsKeyPath: {}, tlsCertPath: {}, threadPool: {}, queueCapacity: {}, " - + "boosLoop: {}, workerLoop: {}, maxInboundMessageSize: {}", - port, threadPoolNums, threadPoolQueueCapacity, - bossLoopNum, workerLoopNum, maxInboundMessageSize); - } - - protected GrpcMessagingProcessor createProcessor() { - return new GrpcMessagingProcessor(grpcForwardService); - } - - protected void configSslContext(NettyServerBuilder serverBuilder) throws SSLException, CertificateException { - if (null == serverBuilder) { - return; - } - ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig(); - boolean tlsTestModeEnable = proxyConfig.isGrpcTlsTestModeEnable(); - if (tlsTestModeEnable) { - SelfSignedCertificate selfSignedCertificate = new SelfSignedCertificate(); - serverBuilder.sslContext(GrpcSslContexts.forServer(selfSignedCertificate.certificate(), selfSignedCertificate.privateKey()) - .trustManager(InsecureTrustManagerFactory.INSTANCE) - .clientAuth(ClientAuth.NONE) - .build()); - return; - } - - String tlsKeyPath = ConfigurationManager.getProxyConfig().getGrpcTlsKeyPath(); - String tlsCertPath = ConfigurationManager.getProxyConfig().getGrpcTlsCertPath(); - try (InputStream serverKeyInputStream = new FileInputStream(tlsKeyPath); - InputStream serverCertificateStream = new FileInputStream(tlsCertPath)) { - serverBuilder.sslContext(GrpcSslContexts.forServer(serverCertificateStream, serverKeyInputStream) - .trustManager(InsecureTrustManagerFactory.INSTANCE) - .clientAuth(ClientAuth.NONE) - .build()); - log.info("TLS configured OK"); - } catch (IOException e) { - log.error("Failed to load Server key/certificate", e); - } - } - - protected void configInterceptor(NettyServerBuilder serverBuilder) { - // grpc interceptors, including acl, logging etc. - if (ConfigurationManager.getProxyConfig().isEnableACL()) { - List accessValidators = ServiceProvider.load(ServiceProvider.ACL_VALIDATOR_ID, AccessValidator.class); - if (accessValidators.isEmpty()) { - throw new IllegalArgumentException("Load AccessValidator failed"); - } - serverBuilder.intercept(new AuthenticationInterceptor(accessValidators)); - } - - serverBuilder.intercept(new ContextInterceptor()) - .intercept(new HeaderInterceptor()); + protected GrpcServer(io.grpc.Server server) { + this.server = server; } public void start() throws Exception { - // first to start grpc service. - this.grpcForwardService.start(); - this.server.start(); log.info("grpc server start successfully."); } @@ -170,10 +40,6 @@ public class GrpcServer implements StartAndShutdown { public void shutdown() { try { this.server.shutdown().awaitTermination(30, TimeUnit.SECONDS); - this.executor.shutdown(); - - this.grpcForwardService.shutdown(); - log.info("grpc server shutdown successfully."); } catch (Exception e) { e.printStackTrace(); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java new file mode 100644 index 0000000000..04cac43db3 --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java @@ -0,0 +1,150 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.rocketmq.proxy.grpc; + +import io.grpc.BindableService; +import io.grpc.ServerServiceDefinition; +import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts; +import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder; +import io.grpc.netty.shaded.io.netty.channel.epoll.EpollEventLoopGroup; +import io.grpc.netty.shaded.io.netty.channel.epoll.EpollServerSocketChannel; +import io.grpc.netty.shaded.io.netty.channel.nio.NioEventLoopGroup; +import io.grpc.netty.shaded.io.netty.channel.socket.nio.NioServerSocketChannel; +import io.grpc.netty.shaded.io.netty.handler.ssl.ClientAuth; +import io.grpc.netty.shaded.io.netty.handler.ssl.util.InsecureTrustManagerFactory; +import io.grpc.netty.shaded.io.netty.handler.ssl.util.SelfSignedCertificate; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.security.cert.CertificateException; +import java.util.List; +import java.util.concurrent.ThreadPoolExecutor; +import javax.net.ssl.SSLException; +import org.apache.rocketmq.acl.AccessValidator; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.common.utils.ServiceProvider; +import org.apache.rocketmq.logging.InternalLogger; +import org.apache.rocketmq.logging.InternalLoggerFactory; +import org.apache.rocketmq.proxy.config.ConfigurationManager; +import org.apache.rocketmq.proxy.config.ProxyConfig; +import org.apache.rocketmq.proxy.grpc.interceptor.AuthenticationInterceptor; +import org.apache.rocketmq.proxy.grpc.interceptor.ContextInterceptor; +import org.apache.rocketmq.proxy.grpc.interceptor.HeaderInterceptor; + +public class GrpcServerBuilder { + private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); + protected NettyServerBuilder serverBuilder; + + public static GrpcServerBuilder newBuilder(ThreadPoolExecutor executor) { + return new GrpcServerBuilder(executor); + } + + public GrpcServerBuilder(ThreadPoolExecutor executor) { + int port = ConfigurationManager.getProxyConfig().getGrpcServerPort(); + serverBuilder = NettyServerBuilder.forPort(port); + + try { + configSslContext(serverBuilder); + } catch (Exception e) { + log.error("grpc tls set failed. msg: {}, e:", e.getMessage(), e); + throw new RuntimeException("grpc tls set failed: " + e.getMessage()); + } + + // build server + int bossLoopNum = ConfigurationManager.getProxyConfig().getGrpcBossLoopNum(); + int workerLoopNum = ConfigurationManager.getProxyConfig().getGrpcWorkerLoopNum(); + int maxInboundMessageSize = ConfigurationManager.getProxyConfig().getGrpcMaxInboundMessageSize(); + + if (ConfigurationManager.getProxyConfig().isEnableGrpcEpoll()) { + serverBuilder.maxInboundMessageSize(maxInboundMessageSize) + .bossEventLoopGroup(new EpollEventLoopGroup(bossLoopNum)) + .workerEventLoopGroup(new EpollEventLoopGroup(workerLoopNum)) + .channelType(EpollServerSocketChannel.class) + .executor(executor); + } else { + serverBuilder.maxInboundMessageSize(maxInboundMessageSize) + .bossEventLoopGroup(new NioEventLoopGroup(bossLoopNum)) + .workerEventLoopGroup(new NioEventLoopGroup(workerLoopNum)) + .channelType(NioServerSocketChannel.class) + .executor(executor); + } + + configInterceptor(serverBuilder); + + log.info( + "grpc server has built. port: {}, tlsKeyPath: {}, tlsCertPath: {}, threadPool: {}, queueCapacity: {}, " + + "boosLoop: {}, workerLoop: {}, maxInboundMessageSize: {}", + port, bossLoopNum, workerLoopNum, maxInboundMessageSize); + } + + public GrpcServerBuilder addService(BindableService service) { + this.serverBuilder.addService(service); + return this; + } + + public GrpcServerBuilder addService(ServerServiceDefinition service) { + this.serverBuilder.addService(service); + return this; + } + + public GrpcServer build() { + return new GrpcServer(this.serverBuilder.build()); + } + + protected void configSslContext(NettyServerBuilder serverBuilder) throws SSLException, CertificateException { + if (null == serverBuilder) { + return; + } + ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig(); + boolean tlsTestModeEnable = proxyConfig.isGrpcTlsTestModeEnable(); + if (tlsTestModeEnable) { + SelfSignedCertificate selfSignedCertificate = new SelfSignedCertificate(); + serverBuilder.sslContext(GrpcSslContexts.forServer(selfSignedCertificate.certificate(), selfSignedCertificate.privateKey()) + .trustManager(InsecureTrustManagerFactory.INSTANCE) + .clientAuth(ClientAuth.NONE) + .build()); + return; + } + + String tlsKeyPath = ConfigurationManager.getProxyConfig().getGrpcTlsKeyPath(); + String tlsCertPath = ConfigurationManager.getProxyConfig().getGrpcTlsCertPath(); + try (InputStream serverKeyInputStream = new FileInputStream(tlsKeyPath); + InputStream serverCertificateStream = new FileInputStream(tlsCertPath)) { + serverBuilder.sslContext(GrpcSslContexts.forServer(serverCertificateStream, serverKeyInputStream) + .trustManager(InsecureTrustManagerFactory.INSTANCE) + .clientAuth(ClientAuth.NONE) + .build()); + log.info("TLS configured OK"); + } catch (IOException e) { + log.error("Failed to load Server key/certificate", e); + } + } + + protected void configInterceptor(NettyServerBuilder serverBuilder) { + // grpc interceptors, including acl, logging etc. + if (ConfigurationManager.getProxyConfig().isEnableACL()) { + List accessValidators = ServiceProvider.load(ServiceProvider.ACL_VALIDATOR_ID, AccessValidator.class); + if (accessValidators.isEmpty()) { + throw new IllegalArgumentException("Load AccessValidator failed"); + } + serverBuilder.intercept(new AuthenticationInterceptor(accessValidators)); + } + + serverBuilder.intercept(new ContextInterceptor()) + .intercept(new HeaderInterceptor()); + } +}