This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 6a158adf3 [ISSUE #4847] HTTPS/(m)TLSv1.3 support for HTTP Admin Server
(#4848)
6a158adf3 is described below
commit 6a158adf344cb525199bee58a5987031e36f7f4f
Author: Pil0tXia <[email protected]>
AuthorDate: Sun Jun 2 20:30:50 2024 +0800
[ISSUE #4847] HTTPS/(m)TLSv1.3 support for HTTP Admin Server (#4848)
* Move admin port to http configs
* Refactor AdminServer to own independent configuration
* Support TLS/SSL for AdminServer
* Add default configs
* Rename jks file to a shorter name
* Add one line of comment
* Fetch current version
---
.../org/apache/eventmesh/common/Constants.java | 2 +
.../common/utils/ConfigurationContextUtil.java | 3 +-
.../config/samples/eventmesh_v1_runtime.yaml | 1 -
eventmesh-runtime/conf/admin-server.jks | Bin 0 -> 2808 bytes
eventmesh-runtime/conf/eventmesh.properties | 24 +++++-
.../admin/handler/v2/ConfigurationHandler.java | 3 +-
.../eventmesh/runtime/boot/AbstractHTTPServer.java | 41 +++-------
.../runtime/boot/AbstractRemotingServer.java | 51 +++---------
.../runtime/boot/EventMeshAdminBootstrap.java | 22 +++++-
.../runtime/boot/EventMeshAdminServer.java | 88 ++++++++++++++++-----
.../runtime/boot/EventMeshGrpcBootstrap.java | 11 +--
.../runtime/boot/EventMeshHTTPServer.java | 66 +---------------
.../runtime/boot/EventMeshHttpBootstrap.java | 15 ++--
.../eventmesh/runtime/boot/EventMeshServer.java | 81 ++++++-------------
.../runtime/boot/EventMeshTcpBootstrap.java | 12 +--
.../runtime/boot/HTTPThreadPoolGroup.java | 51 +++---------
.../eventmesh/runtime/boot/SSLContextFactory.java | 25 +++---
.../configuration/EventMeshAdminConfiguration.java | 61 ++++++++++++++
.../configuration/EventMeshGrpcConfiguration.java | 3 -
.../configuration/EventMeshHTTPConfiguration.java | 4 -
.../configuration/EventMeshTCPConfiguration.java | 3 -
.../http/processor/AdminMetricsProcessor.java | 6 +-
.../runtime/boot/EventMeshServerTest.java | 1 -
.../EventMeshGrpcConfigurationTest.java | 1 -
.../EventMeshHTTPConfigurationTest.java | 1 -
.../EventMeshTCPConfigurationTest.java | 1 -
.../src/test/resources/configuration.properties | 2 -
27 files changed, 259 insertions(+), 320 deletions(-)
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/Constants.java
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/Constants.java
index 97286f096..2460129e7 100644
--- a/eventmesh-common/src/main/java/org/apache/eventmesh/common/Constants.java
+++ b/eventmesh-common/src/main/java/org/apache/eventmesh/common/Constants.java
@@ -200,6 +200,8 @@ public class Constants {
public static final String GRPC = "GRPC";
+ public static final String ADMIN = "ADMIN";
+
public static final String OS_NAME_KEY = "os.name";
public static final String OS_WIN_PREFIX = "win";
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/ConfigurationContextUtil.java
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/ConfigurationContextUtil.java
index 861788899..fede64d65 100644
---
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/ConfigurationContextUtil.java
+++
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/ConfigurationContextUtil.java
@@ -17,6 +17,7 @@
package org.apache.eventmesh.common.utils;
+import static org.apache.eventmesh.common.Constants.ADMIN;
import static org.apache.eventmesh.common.Constants.GRPC;
import static org.apache.eventmesh.common.Constants.HTTP;
import static org.apache.eventmesh.common.Constants.TCP;
@@ -36,7 +37,7 @@ public class ConfigurationContextUtil {
private static final ConcurrentHashMap<String, CommonConfiguration>
CONFIGURATION_MAP = new ConcurrentHashMap<>();
- public static final List<String> KEYS = Lists.newArrayList(HTTP, TCP,
GRPC);
+ public static final List<String> KEYS = Lists.newArrayList(HTTP, TCP,
GRPC, ADMIN);
/**
* Save http, tcp, grpc configuration at startup for global use.
diff --git a/eventmesh-operator/config/samples/eventmesh_v1_runtime.yaml
b/eventmesh-operator/config/samples/eventmesh_v1_runtime.yaml
index fc640aa63..590928ca9 100644
--- a/eventmesh-operator/config/samples/eventmesh_v1_runtime.yaml
+++ b/eventmesh-operator/config/samples/eventmesh_v1_runtime.yaml
@@ -41,7 +41,6 @@ data:
# HTTP Admin Server
eventMesh.server.admin.http.port=10106
########################## eventMesh tcp configuration
############################
- eventMesh.server.tcp.enabled=true
eventMesh.server.tcp.readerIdleSeconds=120
eventMesh.server.tcp.writerIdleSeconds=120
eventMesh.server.tcp.allIdleSeconds=120
diff --git a/eventmesh-runtime/conf/admin-server.jks
b/eventmesh-runtime/conf/admin-server.jks
new file mode 100644
index 000000000..92deb897a
Binary files /dev/null and b/eventmesh-runtime/conf/admin-server.jks differ
diff --git a/eventmesh-runtime/conf/eventmesh.properties
b/eventmesh-runtime/conf/eventmesh.properties
index 4984181ca..9c2e5ee0d 100644
--- a/eventmesh-runtime/conf/eventmesh.properties
+++ b/eventmesh-runtime/conf/eventmesh.properties
@@ -24,11 +24,9 @@ eventMesh.sysid=0000
eventMesh.server.tcp.port=10000
eventMesh.server.http.port=10105
eventMesh.server.grpc.port=10205
-# HTTP Admin Server
eventMesh.server.admin.http.port=10106
-########################## EventMesh TCP Configuration
##########################
-eventMesh.server.tcp.enabled=true
+########################## EventMesh Network Configuration
##########################
eventMesh.server.tcp.readerIdleSeconds=120
eventMesh.server.tcp.writerIdleSeconds=120
eventMesh.server.tcp.allIdleSeconds=120
@@ -64,10 +62,30 @@ eventMesh.server.retry.plugin.type=default
eventMesh.server.gracefulShutdown.sleepIntervalInMills=1000
eventMesh.server.rebalanceRedirect.sleepIntervalInMills=200
+# TLS
+eventMesh.server.useTls.enabled=false
+eventMesh.server.ssl.protocol=TLSv1.1
+eventMesh.server.ssl.cer=sChat2.jks
+eventMesh.server.ssl.pass=sNetty
+
# ip address blacklist
eventMesh.server.blacklist.ipv4=0.0.0.0/8,127.0.0.0/8,169.254.0.0/16,255.255.255.255/32
eventMesh.server.blacklist.ipv6=::/128,::1/128,ff00::/8
+########################## EventMesh HTTP Admin Configuration
##########################
+# thread pool
+eventMesh.server.admin.threads.num=2
+
+# TLS
+eventMesh.server.admin.useTls.enabled=false
+eventMesh.server.admin.ssl.protocol=TLSv1.3
+eventMesh.server.admin.ssl.cer=admin-server.jks
+eventMesh.server.admin.ssl.pass=eventmesh-admin-server
+
+# ip address blacklist
+eventMesh.server.admin.blacklist.ipv4=0.0.0.0/8,127.0.0.0/8,169.254.0.0/16,255.255.255.255/32
+eventMesh.server.admin.blacklist.ipv6=::/128,::1/128,ff00::/8
+
########################## EventMesh Plugin Configuration
##########################
# storage plugin
eventMesh.storage.plugin.type=standalone
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/v2/ConfigurationHandler.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/v2/ConfigurationHandler.java
index 1c149ca52..234e0e79c 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/v2/ConfigurationHandler.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/v2/ConfigurationHandler.java
@@ -27,6 +27,7 @@ import
org.apache.eventmesh.runtime.common.EventMeshHttpHandler;
import org.apache.eventmesh.runtime.configuration.EventMeshGrpcConfiguration;
import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
import org.apache.eventmesh.runtime.configuration.EventMeshTCPConfiguration;
+import org.apache.eventmesh.runtime.constants.EventMeshVersion;
import org.apache.eventmesh.runtime.util.HttpRequestUtil;
import java.lang.reflect.Field;
@@ -126,7 +127,7 @@ public class ConfigurationHandler extends
AbstractHttpHandler {
eventMeshTCPConfiguration,
eventMeshHTTPConfiguration,
eventMeshGrpcConfiguration,
- "v1.10.0-release" // TODO get version number after merging
https://github.com/apache/eventmesh/pull/4055
+ EventMeshVersion.getCurrentVersionDesc()
);
String json =
JSON.toJSONString(Result.success(getConfigurationResponse), filters.toArray(new
Filter[0]));
writeJson(ctx, json);
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
index 97f05a876..3fcca5183 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
@@ -86,6 +86,8 @@ import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import io.opentelemetry.api.trace.Span;
+import lombok.Getter;
+import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
/**
@@ -96,6 +98,8 @@ public abstract class AbstractHTTPServer extends
AbstractRemotingServer {
private final transient EventMeshHTTPConfiguration
eventMeshHttpConfiguration;
+ @Getter
+ @Setter
private EventMeshHttpMetricsManager eventMeshHttpMetricsManager;
private static final DefaultHttpDataFactory DEFAULT_HTTP_DATA_FACTORY =
new DefaultHttpDataFactory(false);
@@ -105,8 +109,14 @@ public abstract class AbstractHTTPServer extends
AbstractRemotingServer {
}
protected final transient AtomicBoolean started = new AtomicBoolean(false);
+
+ @Getter
private final transient boolean useTLS;
+
+ @Getter
+ @Setter
private Boolean useTrace = false; // Determine whether trace is enabled
+
private static final int MAX_CONNECTIONS = 20_000;
/**
@@ -118,10 +128,13 @@ public abstract class AbstractHTTPServer extends
AbstractRemotingServer {
private HttpConnectionHandler httpConnectionHandler;
private HttpDispatcher httpDispatcher;
+ @Setter
+ @Getter
private HandlerService handlerService;
private final transient ThreadPoolExecutor asyncContextCompleteHandler =
ThreadPoolFactory.createThreadPoolExecutor(10, 10,
"EventMesh-http-asyncContext");
+ @Getter
private final HTTPThreadPoolGroup httpThreadPoolGroup;
public AbstractHTTPServer(final int port, final boolean useTLS,
@@ -523,32 +536,4 @@ public abstract class AbstractHTTPServer extends
AbstractRemotingServer {
httpDispatcher);
}
}
-
- public void setUseTrace(final Boolean useTrace) {
- this.useTrace = useTrace;
- }
-
- public Boolean getUseTrace() {
- return useTrace;
- }
-
- public EventMeshHttpMetricsManager getEventMeshHttpMetricsManager() {
- return eventMeshHttpMetricsManager;
- }
-
- public void setEventMeshHttpMetricsManager(final
EventMeshHttpMetricsManager eventMeshHttpMetricsManager) {
- this.eventMeshHttpMetricsManager = eventMeshHttpMetricsManager;
- }
-
- public HTTPThreadPoolGroup getHttpThreadPoolGroup() {
- return httpThreadPoolGroup;
- }
-
- public HandlerService getHandlerService() {
- return handlerService;
- }
-
- public void setHandlerService(HandlerService handlerService) {
- this.handlerService = handlerService;
- }
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractRemotingServer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractRemotingServer.java
index fb0911662..e02637ec3 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractRemotingServer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractRemotingServer.java
@@ -22,7 +22,6 @@ import org.apache.eventmesh.common.utils.SystemUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.runtime.core.protocol.producer.ProducerManager;
-import java.util.Objects;
import java.util.concurrent.TimeUnit;
import io.netty.channel.EventLoopGroup;
@@ -31,22 +30,32 @@ import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.EventExecutorGroup;
+import lombok.Getter;
+import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
/**
* The most basic server
*/
@Slf4j
+@Getter
public abstract class AbstractRemotingServer implements RemotingServer {
private static final int MAX_THREADS =
Runtime.getRuntime().availableProcessors();
private static final int DEFAULT_SLEEP_SECONDS = 30;
+ @Setter
private EventLoopGroup bossGroup;
+
+ @Setter
private EventLoopGroup ioGroup;
+
+ @Setter
private EventExecutorGroup workerGroup;
+
protected ProducerManager producerManager;
+ @Setter
private int port;
protected void buildBossGroup(final String threadPrefix) {
@@ -75,10 +84,6 @@ public abstract class AbstractRemotingServer implements
RemotingServer {
producerManager.init();
}
- public ProducerManager getProducerManager() {
- return producerManager;
- }
-
public void init(final String threadPrefix) throws Exception {
buildBossGroup(threadPrefix);
buildIOGroup(threadPrefix);
@@ -94,16 +99,16 @@ public abstract class AbstractRemotingServer implements
RemotingServer {
bossGroup.shutdownGracefully();
log.info("shutdown bossGroup");
}
- if (Objects.isNull(producerManager)) {
+ if (producerManager != null) {
producerManager.shutdown();
}
+
ThreadUtils.randomPause(TimeUnit.SECONDS.toMillis(DEFAULT_SLEEP_SECONDS));
if (ioGroup != null) {
ioGroup.shutdownGracefully();
log.info("shutdown ioGroup");
}
-
if (workerGroup != null) {
workerGroup.shutdownGracefully();
@@ -114,36 +119,4 @@ public abstract class AbstractRemotingServer implements
RemotingServer {
protected boolean useEpoll() {
return SystemUtils.isLinuxPlatform() && Epoll.isAvailable();
}
-
- public EventLoopGroup getBossGroup() {
- return bossGroup;
- }
-
- public void setBossGroup(final EventLoopGroup bossGroup) {
- this.bossGroup = bossGroup;
- }
-
- public EventLoopGroup getIoGroup() {
- return ioGroup;
- }
-
- public void setIoGroup(final EventLoopGroup ioGroup) {
- this.ioGroup = ioGroup;
- }
-
- public EventExecutorGroup getWorkerGroup() {
- return workerGroup;
- }
-
- public void setWorkerGroup(final EventExecutorGroup workerGroup) {
- this.workerGroup = workerGroup;
- }
-
- public int getPort() {
- return port;
- }
-
- public void setPort(final int port) {
- this.port = port;
- }
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshAdminBootstrap.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshAdminBootstrap.java
index e125475ec..b170afd3a 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshAdminBootstrap.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshAdminBootstrap.java
@@ -17,23 +17,38 @@
package org.apache.eventmesh.runtime.boot;
+import static org.apache.eventmesh.common.Constants.ADMIN;
+
+import org.apache.eventmesh.common.config.ConfigService;
+import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
+import org.apache.eventmesh.runtime.configuration.EventMeshAdminConfiguration;
+
+import lombok.Getter;
+
public class EventMeshAdminBootstrap implements EventMeshBootstrap {
+ @Getter
private EventMeshAdminServer eventMeshAdminServer;
- private EventMeshServer eventMeshServer;
+ private final EventMeshAdminConfiguration eventMeshAdminConfiguration;
+
+ private final EventMeshServer eventMeshServer;
public EventMeshAdminBootstrap(EventMeshServer eventMeshServer) {
this.eventMeshServer = eventMeshServer;
+
+ ConfigService configService = ConfigService.getInstance();
+ this.eventMeshAdminConfiguration =
configService.buildConfigInstance(EventMeshAdminConfiguration.class);
+
+ ConfigurationContextUtil.putIfAbsent(ADMIN,
eventMeshAdminConfiguration);
}
@Override
public void init() throws Exception {
if (eventMeshServer != null) {
- eventMeshAdminServer = new EventMeshAdminServer(eventMeshServer);
+ eventMeshAdminServer = new EventMeshAdminServer(eventMeshServer,
eventMeshAdminConfiguration);
eventMeshAdminServer.init();
}
-
}
@Override
@@ -41,7 +56,6 @@ public class EventMeshAdminBootstrap implements
EventMeshBootstrap {
if (eventMeshAdminServer != null) {
eventMeshAdminServer.start();
}
-
}
@Override
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshAdminServer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshAdminServer.java
index 4abba74a5..5e98fc690 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshAdminServer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshAdminServer.java
@@ -17,13 +17,22 @@
package org.apache.eventmesh.runtime.boot;
+import org.apache.eventmesh.common.ThreadPoolFactory;
+import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.runtime.admin.handler.AdminHandlerManager;
import org.apache.eventmesh.runtime.admin.handler.HttpHandler;
+import org.apache.eventmesh.runtime.configuration.EventMeshAdminConfiguration;
+import
org.apache.eventmesh.runtime.core.protocol.http.processor.AdminMetricsProcessor;
import org.apache.eventmesh.runtime.util.HttpResponseUtils;
import java.net.URI;
import java.util.Objects;
import java.util.Optional;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFutureListener;
@@ -41,27 +50,37 @@ import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.HttpRequestDecoder;
import io.netty.handler.codec.http.HttpResponseEncoder;
import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.ssl.SslHandler;
+import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class EventMeshAdminServer extends AbstractHTTPServer {
+ private final EventMeshAdminConfiguration eventMeshAdminConfiguration;
+
private HttpConnectionHandler httpConnectionHandler = new
HttpConnectionHandler();
private AdminHandlerManager adminHandlerManager;
- public EventMeshAdminServer(EventMeshServer eventMeshServer) {
-
super(eventMeshServer.getEventMeshTCPServer().getEventMeshTCPConfiguration().getEventMeshServerAdminPort(),
false,
-
eventMeshServer.getEventMeshHTTPServer().getEventMeshHttpConfiguration());
+ @Getter
+ private ThreadPoolExecutor adminMetricsExecutor;
+
+ public EventMeshAdminServer(final EventMeshServer eventMeshServer, final
EventMeshAdminConfiguration eventMeshAdminConfiguration) {
+ super(eventMeshAdminConfiguration.getEventMeshServerAdminPort(),
+ eventMeshAdminConfiguration.isEventMeshServerUseTls(),
+ eventMeshAdminConfiguration);
+ this.eventMeshAdminConfiguration = eventMeshAdminConfiguration;
adminHandlerManager = new AdminHandlerManager(eventMeshServer);
}
-
@Override
public void init() throws Exception {
super.init("eventMesh-admin");
+ initThreadPool();
adminHandlerManager.registerHttpHandler();
+ registerAdminRequestProcessor();
}
@Override
@@ -71,8 +90,9 @@ public class EventMeshAdminServer extends AbstractHTTPServer {
try {
bootstrap.group(this.getBossGroup(), this.getIoGroup())
.channel(useEpoll() ? EpollServerSocketChannel.class :
NioServerSocketChannel.class)
- .childHandler(new AdminServerInitializer())
- .childOption(ChannelOption.SO_KEEPALIVE, Boolean.TRUE);
+ .childHandler(new AdminServerInitializer(
+ this.isUseTLS() ?
SSLContextFactory.getSslContext(eventMeshAdminConfiguration) : null,
this.isUseTLS()))
+ .childOption(ChannelOption.AUTO_CLOSE, Boolean.TRUE);
log.info("AdminHttpServer[port={}] started.", this.getPort());
@@ -95,29 +115,38 @@ public class EventMeshAdminServer extends
AbstractHTTPServer {
started.compareAndSet(false, true);
}
- public void parseHttpRequest(ChannelHandlerContext ctx, HttpRequest
httpRequest) {
- String uriStr = httpRequest.uri();
- URI uri = URI.create(uriStr);
- Optional<HttpHandler> httpHandlerOpt =
adminHandlerManager.getHttpHandler(uri.getPath());
- if (httpHandlerOpt.isPresent()) {
- try {
- httpHandlerOpt.get().handle(httpRequest, ctx);
- } catch (Exception e) {
- log.error("admin server channelRead error", e);
-
ctx.writeAndFlush(HttpResponseUtils.buildHttpResponse(Objects.requireNonNull(e.getMessage()),
ctx,
- HttpHeaderValues.APPLICATION_JSON,
HttpResponseStatus.INTERNAL_SERVER_ERROR)).addListener(ChannelFutureListener.CLOSE);
- }
- } else {
-
ctx.writeAndFlush(HttpResponseUtils.createNotFound()).addListener(ChannelFutureListener.CLOSE);
- }
+ private void initThreadPool() {
+ adminMetricsExecutor = ThreadPoolFactory.createThreadPoolExecutor(
+ eventMeshAdminConfiguration.getEventMeshServerAdminThreadNum(),
+ eventMeshAdminConfiguration.getEventMeshServerAdminThreadNum(),
+ new LinkedBlockingQueue<>(50), "eventMesh-admin-metrics", true);
+ }
+
+ private void registerAdminRequestProcessor() {
+ final AdminMetricsProcessor adminMetricsProcessor = new
AdminMetricsProcessor(this);
+ registerProcessor(RequestCode.ADMIN_METRICS.getRequestCode(),
adminMetricsProcessor);
}
private class AdminServerInitializer extends
ChannelInitializer<SocketChannel> {
+ private final transient SSLContext sslContext;
+ private final transient boolean useTLS;
+
+ public AdminServerInitializer(final SSLContext sslContext, final
boolean useTLS) {
+ this.sslContext = sslContext;
+ this.useTLS = useTLS;
+ }
+
@Override
protected void initChannel(final SocketChannel channel) {
final ChannelPipeline pipeline = channel.pipeline();
+ if (sslContext != null && useTLS) {
+ final SSLEngine sslEngine = sslContext.createSSLEngine();
+ sslEngine.setUseClientMode(false);
+ pipeline.addFirst(getWorkerGroup(), "ssl", new
SslHandler(sslEngine));
+ }
+
pipeline.addLast(getWorkerGroup(),
new HttpRequestDecoder(),
new HttpResponseEncoder(),
@@ -131,5 +160,22 @@ public class EventMeshAdminServer extends
AbstractHTTPServer {
}
});
}
+
+ private void parseHttpRequest(ChannelHandlerContext ctx, HttpRequest
httpRequest) {
+ String uriStr = httpRequest.uri();
+ URI uri = URI.create(uriStr);
+ Optional<HttpHandler> httpHandlerOpt =
adminHandlerManager.getHttpHandler(uri.getPath());
+ if (httpHandlerOpt.isPresent()) {
+ try {
+ httpHandlerOpt.get().handle(httpRequest, ctx);
+ } catch (Exception e) {
+ log.error("admin server channelRead error", e);
+
ctx.writeAndFlush(HttpResponseUtils.buildHttpResponse(Objects.requireNonNull(e.getMessage()),
ctx,
+ HttpHeaderValues.APPLICATION_JSON,
HttpResponseStatus.INTERNAL_SERVER_ERROR)).addListener(ChannelFutureListener.CLOSE);
+ }
+ } else {
+
ctx.writeAndFlush(HttpResponseUtils.createNotFound()).addListener(ChannelFutureListener.CLOSE);
+ }
+ }
}
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshGrpcBootstrap.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshGrpcBootstrap.java
index dc218084b..b6e493c1b 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshGrpcBootstrap.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshGrpcBootstrap.java
@@ -23,10 +23,13 @@ import org.apache.eventmesh.common.config.ConfigService;
import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
import org.apache.eventmesh.runtime.configuration.EventMeshGrpcConfiguration;
+import lombok.Getter;
+
public class EventMeshGrpcBootstrap implements EventMeshBootstrap {
private final EventMeshGrpcConfiguration eventMeshGrpcConfiguration;
+ @Getter
private EventMeshGrpcServer eventMeshGrpcServer;
private final EventMeshServer eventMeshServer;
@@ -62,12 +65,4 @@ public class EventMeshGrpcBootstrap implements
EventMeshBootstrap {
eventMeshGrpcServer.shutdown();
}
}
-
- public EventMeshGrpcServer getEventMeshGrpcServer() {
- return eventMeshGrpcServer;
- }
-
- public void setEventMeshGrpcServer(EventMeshGrpcServer
eventMeshGrpcServer) {
- this.eventMeshGrpcServer = eventMeshGrpcServer;
- }
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
index cc47cc972..1089a1caf 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
@@ -31,7 +31,6 @@ import
org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
import org.apache.eventmesh.runtime.core.consumer.SubscriptionManager;
import
org.apache.eventmesh.runtime.core.protocol.http.consumer.ConsumerManager;
-import
org.apache.eventmesh.runtime.core.protocol.http.processor.AdminMetricsProcessor;
import
org.apache.eventmesh.runtime.core.protocol.http.processor.BatchSendMessageProcessor;
import
org.apache.eventmesh.runtime.core.protocol.http.processor.BatchSendMessageV2Processor;
import
org.apache.eventmesh.runtime.core.protocol.http.processor.CreateTopicProcessor;
@@ -68,6 +67,7 @@ import org.assertj.core.util.Lists;
import com.google.common.eventbus.EventBus;
import com.google.common.util.concurrent.RateLimiter;
+import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
@@ -75,6 +75,7 @@ import lombok.extern.slf4j.Slf4j;
* Add multiple managers to the underlying server
*/
@Slf4j
+@Getter
public class EventMeshHTTPServer extends AbstractHTTPServer {
private final EventMeshServer eventMeshServer;
@@ -95,7 +96,6 @@ public class EventMeshHTTPServer extends AbstractHTTPServer {
private transient RateLimiter batchRateLimiter;
public EventMeshHTTPServer(final EventMeshServer eventMeshServer, final
EventMeshHTTPConfiguration eventMeshHttpConfiguration) {
-
super(eventMeshHttpConfiguration.getHttpServerPort(),
eventMeshHttpConfiguration.isEventMeshServerUseTls(),
eventMeshHttpConfiguration);
@@ -103,7 +103,6 @@ public class EventMeshHTTPServer extends AbstractHTTPServer
{
this.eventMeshHttpConfiguration = eventMeshHttpConfiguration;
this.metaStorage = eventMeshServer.getMetaStorage();
this.acl = eventMeshServer.getAcl();
-
}
public void init() throws Exception {
@@ -251,9 +250,6 @@ public class EventMeshHTTPServer extends AbstractHTTPServer
{
final SendAsyncRemoteEventProcessor sendAsyncRemoteEventProcessor =
new SendAsyncRemoteEventProcessor(this);
this.getHandlerService().register(sendAsyncRemoteEventProcessor);
- final AdminMetricsProcessor adminMetricsProcessor = new
AdminMetricsProcessor(this);
- registerProcessor(RequestCode.ADMIN_METRICS.getRequestCode(),
adminMetricsProcessor);
-
final HeartBeatProcessor heartProcessor = new HeartBeatProcessor(this);
registerProcessor(RequestCode.HEARTBEAT.getRequestCode(),
heartProcessor);
@@ -299,62 +295,4 @@ public class EventMeshHTTPServer extends
AbstractHTTPServer {
this.getHandlerService().register(webHookProcessor,
super.getHttpThreadPoolGroup().getWebhookExecutor());
}
-
- public SubscriptionManager getSubscriptionManager() {
- return subscriptionManager;
- }
-
- public ConsumerManager getConsumerManager() {
- return consumerManager;
- }
-
- public ProducerManager getProducerManager() {
- return producerManager;
- }
-
- public EventMeshHTTPConfiguration getEventMeshHttpConfiguration() {
- return eventMeshHttpConfiguration;
- }
-
- public EventBus getEventBus() {
- return eventBus;
- }
-
- public HttpRetryer getHttpRetryer() {
- return httpRetryer;
- }
-
- public Acl getAcl() {
- return acl;
- }
-
- public EventMeshServer getEventMeshServer() {
- return eventMeshServer;
- }
-
- public RateLimiter getMsgRateLimiter() {
- return msgRateLimiter;
- }
-
- public RateLimiter getBatchRateLimiter() {
- return batchRateLimiter;
- }
-
- public FilterEngine getFilterEngine() {
- return filterEngine;
- }
-
- public TransformerEngine getTransformerEngine() {
- return transformerEngine;
- }
-
- public MetaStorage getMetaStorage() {
- return metaStorage;
- }
-
- public HTTPClientPool getHttpClientPool() {
- return httpClientPool;
- }
-
-
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHttpBootstrap.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHttpBootstrap.java
index 87ce3c23a..df3b227e4 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHttpBootstrap.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHttpBootstrap.java
@@ -23,12 +23,15 @@ import org.apache.eventmesh.common.config.ConfigService;
import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
-public class EventMeshHttpBootstrap implements EventMeshBootstrap {
+import lombok.Getter;
- private final EventMeshHTTPConfiguration eventMeshHttpConfiguration;
+public class EventMeshHttpBootstrap implements EventMeshBootstrap {
+ @Getter
public EventMeshHTTPServer eventMeshHttpServer;
+ private final EventMeshHTTPConfiguration eventMeshHttpConfiguration;
+
private final EventMeshServer eventMeshServer;
public EventMeshHttpBootstrap(final EventMeshServer eventMeshServer) {
@@ -64,12 +67,4 @@ public class EventMeshHttpBootstrap implements
EventMeshBootstrap {
eventMeshHttpServer.shutdown();
}
}
-
- public EventMeshHTTPServer getEventMeshHttpServer() {
- return eventMeshHttpServer;
- }
-
- public void setEventMeshHttpServer(EventMeshHTTPServer
eventMeshHttpServer) {
- this.eventMeshHttpServer = eventMeshHttpServer;
- }
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
index b3250795d..d61580b9c 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
@@ -43,23 +43,32 @@ import java.util.Objects;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.stream.Collectors;
+import lombok.Getter;
+import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class EventMeshServer {
+ @Getter
private final Acl acl;
+ @Getter
+ @Setter
private MetaStorage metaStorage;
+ @Getter
private static Trace trace;
private final StorageResource storageResource;
+ @Getter
private ServiceState serviceState;
+ @Getter
private ProducerTopicManager producerTopicManager;
+ @Getter
private final CommonConfiguration configuration;
// private transient ClientManageController clientManageController;
@@ -70,13 +79,17 @@ public class EventMeshServer {
private static final ConfigService configService =
ConfigService.getInstance();
- private EventMeshAdminBootstrap adminBootstrap;
-
+ @Getter
private EventMeshTCPServer eventMeshTCPServer = null;
+ @Getter
+ private EventMeshHTTPServer eventMeshHTTPServer = null;
+
+ @Getter
private EventMeshGrpcServer eventMeshGrpcServer = null;
- private EventMeshHTTPServer eventMeshHTTPServer = null;
+ @Getter
+ private EventMeshAdminServer eventMeshAdminServer = null;
private EventMeshMetricsManager eventMeshMetricsManager;
@@ -105,8 +118,7 @@ public class EventMeshServer {
case GRPC:
BOOTSTRAP_LIST.add(new EventMeshGrpcBootstrap(this));
break;
- default:
- // nothing to do
+ default: // nothing to do
}
}
@@ -115,6 +127,9 @@ public class EventMeshServer {
BOOTSTRAP_LIST.add(new EventMeshTcpBootstrap(this));
}
+ // HTTP Admin Server always enabled
+ BOOTSTRAP_LIST.add(new EventMeshAdminBootstrap(this));
+
List<String> metricsPluginTypes =
configuration.getEventMeshMetricsPluginType();
if (CollectionUtils.isNotEmpty(metricsPluginTypes)) {
List<MetricsRegistry> metricsRegistries =
metricsPluginTypes.stream().map(metric ->
MetricsPluginFactory.getMetricsRegistry(metric))
@@ -147,6 +162,9 @@ public class EventMeshServer {
if (eventMeshBootstrap instanceof EventMeshGrpcBootstrap) {
eventMeshGrpcServer = ((EventMeshGrpcBootstrap)
eventMeshBootstrap).getEventMeshGrpcServer();
}
+ if (eventMeshBootstrap instanceof EventMeshAdminBootstrap) {
+ eventMeshAdminServer = ((EventMeshAdminBootstrap)
eventMeshBootstrap).getEventMeshAdminServer();
+ }
}
if (Objects.nonNull(eventMeshTCPServer)) {
@@ -168,12 +186,6 @@ public class EventMeshServer {
eventMeshMetricsManager.init();
}
-
- if (Objects.nonNull(eventMeshTCPServer) &&
Objects.nonNull(eventMeshHTTPServer) && Objects.nonNull(eventMeshGrpcServer)) {
- adminBootstrap = new EventMeshAdminBootstrap(this);
- adminBootstrap.init();
- }
-
producerTopicManager = new ProducerTopicManager(this);
producerTopicManager.init();
@@ -203,13 +215,10 @@ public class EventMeshServer {
eventMeshBootstrap.start();
}
- if (Objects.nonNull(adminBootstrap)) {
- adminBootstrap.start();
- }
producerTopicManager.start();
+
serviceState = ServiceState.RUNNING;
log.info(SERVER_STATE_MSG, serviceState);
-
}
public void shutdown() throws Exception {
@@ -235,48 +244,8 @@ public class EventMeshServer {
}
producerTopicManager.shutdown();
ConfigurationContextUtil.clear();
- serviceState = ServiceState.STOPPED;
+ serviceState = ServiceState.STOPPED;
log.info(SERVER_STATE_MSG, serviceState);
}
-
- public static Trace getTrace() {
- return trace;
- }
-
- public ServiceState getServiceState() {
- return serviceState;
- }
-
- public MetaStorage getMetaStorage() {
- return metaStorage;
- }
-
- public void setMetaStorage(final MetaStorage metaStorage) {
- this.metaStorage = metaStorage;
- }
-
- public Acl getAcl() {
- return acl;
- }
-
- public ProducerTopicManager getProducerTopicManager() {
- return producerTopicManager;
- }
-
- public CommonConfiguration getConfiguration() {
- return configuration;
- }
-
- public EventMeshTCPServer getEventMeshTCPServer() {
- return eventMeshTCPServer;
- }
-
- public EventMeshGrpcServer getEventMeshGrpcServer() {
- return eventMeshGrpcServer;
- }
-
- public EventMeshHTTPServer getEventMeshHTTPServer() {
- return eventMeshHTTPServer;
- }
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTcpBootstrap.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTcpBootstrap.java
index 9cd665d28..e098b203c 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTcpBootstrap.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTcpBootstrap.java
@@ -23,8 +23,11 @@ import org.apache.eventmesh.common.config.ConfigService;
import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
import org.apache.eventmesh.runtime.configuration.EventMeshTCPConfiguration;
+import lombok.Getter;
+
public class EventMeshTcpBootstrap implements EventMeshBootstrap {
+ @Getter
private EventMeshTCPServer eventMeshTcpServer;
private final EventMeshTCPConfiguration eventMeshTcpConfiguration;
@@ -63,13 +66,4 @@ public class EventMeshTcpBootstrap implements
EventMeshBootstrap {
eventMeshTcpServer.shutdown();
}
}
-
- public EventMeshTCPServer getEventMeshTcpServer() {
- return eventMeshTcpServer;
- }
-
- public void setEventMeshTcpServer(EventMeshTCPServer eventMeshTcpServer) {
- this.eventMeshTcpServer = eventMeshTcpServer;
- }
-
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/HTTPThreadPoolGroup.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/HTTPThreadPoolGroup.java
index bf6c740b5..aa1178850 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/HTTPThreadPoolGroup.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/HTTPThreadPoolGroup.java
@@ -23,17 +23,25 @@ import
org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
+import lombok.Getter;
+
public class HTTPThreadPoolGroup implements ThreadPoolGroup {
private final EventMeshHTTPConfiguration eventMeshHttpConfiguration;
+ @Getter
private ThreadPoolExecutor batchMsgExecutor;
+ @Getter
private ThreadPoolExecutor sendMsgExecutor;
+ @Getter
private ThreadPoolExecutor remoteMsgExecutor;
+ @Getter
private ThreadPoolExecutor replyMsgExecutor;
+ @Getter
private ThreadPoolExecutor pushMsgExecutor;
+ @Getter
private ThreadPoolExecutor clientManageExecutor;
- private ThreadPoolExecutor runtimeAdminExecutor;
+ @Getter
private ThreadPoolExecutor webhookExecutor;
public HTTPThreadPoolGroup(EventMeshHTTPConfiguration
eventMeshHttpConfiguration) {
@@ -73,12 +81,6 @@ public class HTTPThreadPoolGroup implements ThreadPoolGroup {
new
LinkedBlockingQueue<>(eventMeshHttpConfiguration.getEventMeshServerClientManageBlockQSize()),
"eventMesh-clientManage", true);
- // The runtimeAdminExecutor here is for the runtime.admin package.
- runtimeAdminExecutor = ThreadPoolFactory.createThreadPoolExecutor(
- eventMeshHttpConfiguration.getEventMeshServerAdminThreadNum(),
- eventMeshHttpConfiguration.getEventMeshServerAdminThreadNum(),
- new LinkedBlockingQueue<>(50), "eventMesh-runtime-admin", true);
-
replyMsgExecutor = ThreadPoolFactory.createThreadPoolExecutor(
eventMeshHttpConfiguration.getEventMeshServerReplyMsgThreadNum(),
eventMeshHttpConfiguration.getEventMeshServerReplyMsgThreadNum(),
@@ -95,9 +97,6 @@ public class HTTPThreadPoolGroup implements ThreadPoolGroup {
if (batchMsgExecutor != null) {
batchMsgExecutor.shutdown();
}
- if (runtimeAdminExecutor != null) {
- runtimeAdminExecutor.shutdown();
- }
if (clientManageExecutor != null) {
clientManageExecutor.shutdown();
}
@@ -114,36 +113,4 @@ public class HTTPThreadPoolGroup implements
ThreadPoolGroup {
replyMsgExecutor.shutdown();
}
}
-
- public ThreadPoolExecutor getBatchMsgExecutor() {
- return batchMsgExecutor;
- }
-
- public ThreadPoolExecutor getSendMsgExecutor() {
- return sendMsgExecutor;
- }
-
- public ThreadPoolExecutor getRemoteMsgExecutor() {
- return remoteMsgExecutor;
- }
-
- public ThreadPoolExecutor getReplyMsgExecutor() {
- return replyMsgExecutor;
- }
-
- public ThreadPoolExecutor getPushMsgExecutor() {
- return pushMsgExecutor;
- }
-
- public ThreadPoolExecutor getClientManageExecutor() {
- return clientManageExecutor;
- }
-
- public ThreadPoolExecutor getRuntimeAdminExecutor() {
- return runtimeAdminExecutor;
- }
-
- public ThreadPoolExecutor getWebhookExecutor() {
- return webhookExecutor;
- }
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/SSLContextFactory.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/SSLContextFactory.java
index 0f48220a4..a0736b430 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/SSLContextFactory.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/SSLContextFactory.java
@@ -17,6 +17,7 @@
package org.apache.eventmesh.runtime.boot;
+import org.apache.eventmesh.runtime.configuration.EventMeshAdminConfiguration;
import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
@@ -40,23 +41,19 @@ import javax.net.ssl.SSLContext;
public class SSLContextFactory {
- private static String protocol = "TLSv1.1";
-
- private static String fileName;
-
- private static String password;
-
+ /**
+ * {@link EventMeshAdminConfiguration} will be parsed into {@link
EventMeshHTTPConfiguration}.
+ */
public static SSLContext getSslContext(final EventMeshHTTPConfiguration
eventMeshHttpConfiguration)
- throws NoSuchAlgorithmException, KeyStoreException,
CertificateException, IOException,
- UnrecoverableKeyException, KeyManagementException {
+ throws NoSuchAlgorithmException, KeyStoreException,
CertificateException, IOException, UnrecoverableKeyException,
KeyManagementException {
+
+ String protocol =
eventMeshHttpConfiguration.getEventMeshServerSSLProtocol();
+ String fileName =
eventMeshHttpConfiguration.getEventMeshServerSSLCer();
+ String password =
eventMeshHttpConfiguration.getEventMeshServerSSLPass();
SSLContext sslContext;
- try (InputStream inputStream =
Files.newInputStream(Paths.get(EventMeshConstants.EVENTMESH_CONF_HOME
- + File.separator
- + fileName), StandardOpenOption.READ)) {
- protocol =
eventMeshHttpConfiguration.getEventMeshServerSSLProtocol();
- fileName = eventMeshHttpConfiguration.getEventMeshServerSSLCer();
- password = eventMeshHttpConfiguration.getEventMeshServerSSLPass();
+ try (InputStream inputStream =
Files.newInputStream(Paths.get(EventMeshConstants.EVENTMESH_CONF_HOME +
File.separator + fileName),
+ StandardOpenOption.READ)) {
char[] filePass = StringUtils.isNotBlank(password) ?
password.toCharArray() : new char[0];
final KeyStore keyStore = KeyStore.getInstance("JKS");
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshAdminConfiguration.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshAdminConfiguration.java
new file mode 100644
index 000000000..dff80eaaa
--- /dev/null
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshAdminConfiguration.java
@@ -0,0 +1,61 @@
+/*
+ * 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.eventmesh.runtime.configuration;
+
+import org.apache.eventmesh.common.config.Config;
+import org.apache.eventmesh.common.config.ConfigField;
+
+import java.util.Collections;
+import java.util.List;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+
+import inet.ipaddr.IPAddress;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+@NoArgsConstructor
+@Config(prefix = "eventMesh.server")
+public class EventMeshAdminConfiguration extends EventMeshHTTPConfiguration {
+
+ @ConfigField(field = "admin.http.port")
+ private int eventMeshServerAdminPort = 10106;
+
+ @ConfigField(field = "admin.threads.num")
+ private int eventMeshServerAdminThreadNum = 2;
+
+ @ConfigField(field = "admin.useTls.enabled")
+ private boolean eventMeshServerUseTls = false;
+
+ @ConfigField(field = "admin.ssl.protocol")
+ private String eventMeshServerSSLProtocol = "TLSv1.3";
+
+ @ConfigField(field = "admin.ssl.cer")
+ private String eventMeshServerSSLCer = "admin-server.jks";
+
+ @ConfigField(field = "admin.ssl.pass")
+ private String eventMeshServerSSLPass = "eventmesh-admin-server";
+
+ @ConfigField(field = "admin.blacklist.ipv4")
+ private List<IPAddress> eventMeshIpv4BlackList = Collections.emptyList();
// TODO implement after merging #4835
+
+ @ConfigField(field = "admin.blacklist.ipv6")
+ private List<IPAddress> eventMeshIpv6BlackList = Collections.emptyList();
+}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshGrpcConfiguration.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshGrpcConfiguration.java
index cf08f2755..924a07ab0 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshGrpcConfiguration.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshGrpcConfiguration.java
@@ -59,9 +59,6 @@ public class EventMeshGrpcConfiguration extends
CommonConfiguration {
@ConfigField(field = "metaStorage.threads.num")
private int eventMeshServerMetaStorageThreadNum = 10;
- @ConfigField(field = "admin.threads.num")
- private int eventMeshServerAdminThreadNum = 2;
-
@ConfigField(field = "retry.threads.num")
private int eventMeshServerRetryThreadNum = 2;
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshHTTPConfiguration.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshHTTPConfiguration.java
index b0fa18df4..287c22224 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshHTTPConfiguration.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshHTTPConfiguration.java
@@ -63,9 +63,6 @@ public class EventMeshHTTPConfiguration extends
CommonConfiguration {
@ConfigField(field = "metaStorage.threads.num")
private int eventMeshServerMetaStorageThreadNum = 10;
- @ConfigField(field = "admin.threads.num")
- private int eventMeshServerAdminThreadNum = 2;
-
@ConfigField(field = "retry.threads.num")
private int eventMeshServerRetryThreadNum = 2;
@@ -131,5 +128,4 @@ public class EventMeshHTTPConfiguration extends
CommonConfiguration {
@ConfigField(field = "blacklist.ipv6")
private List<IPAddress> eventMeshIpv6BlackList = Collections.emptyList();
-
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshTCPConfiguration.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshTCPConfiguration.java
index 891f98616..907d80f68 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshTCPConfiguration.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshTCPConfiguration.java
@@ -109,9 +109,6 @@ public class EventMeshTCPConfiguration extends
CommonConfiguration {
@ConfigField(field = "tcp.RebalanceIntervalInMills")
private Integer eventMeshTcpRebalanceIntervalInMills = 30 * 1000;
- @ConfigField(field = "admin.http.port")
- private int eventMeshServerAdminPort = 10106;
-
@ConfigField(field = "tcp.sendBack.enabled")
private boolean eventMeshTcpSendBackEnabled = Boolean.TRUE;
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminMetricsProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminMetricsProcessor.java
index b3fbf0d6a..9a8b36934 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminMetricsProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminMetricsProcessor.java
@@ -18,7 +18,7 @@
package org.apache.eventmesh.runtime.core.protocol.http.processor;
import org.apache.eventmesh.common.protocol.http.HttpCommand;
-import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
+import org.apache.eventmesh.runtime.boot.EventMeshAdminServer;
import org.apache.eventmesh.runtime.core.protocol.http.async.AsyncContext;
import java.util.concurrent.Executor;
@@ -31,7 +31,7 @@ import lombok.RequiredArgsConstructor;
@RequiredArgsConstructor
public class AdminMetricsProcessor extends AbstractHttpRequestProcessor {
- private final EventMeshHTTPServer eventMeshHTTPServer;
+ private final EventMeshAdminServer eventMeshAdminServer;
@Override
public void processRequest(ChannelHandlerContext ctx,
AsyncContext<HttpCommand> asyncContext) throws Exception {
@@ -39,6 +39,6 @@ public class AdminMetricsProcessor extends
AbstractHttpRequestProcessor {
@Override
public Executor executor() {
- return
eventMeshHTTPServer.getHttpThreadPoolGroup().getRuntimeAdminExecutor();
+ return eventMeshAdminServer.getAdminMetricsExecutor();
}
}
diff --git
a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/boot/EventMeshServerTest.java
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/boot/EventMeshServerTest.java
index c546d38e3..83baf20fb 100644
---
a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/boot/EventMeshServerTest.java
+++
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/boot/EventMeshServerTest.java
@@ -77,7 +77,6 @@ public class EventMeshServerTest {
Assertions.assertEquals(15816,
config.getEventMeshTcpMsgRetryAsyncDelayInMills());
Assertions.assertEquals(16816,
config.getEventMeshTcpMsgRetryQueueSize());
Assertions.assertEquals(Integer.valueOf(17816),
config.getEventMeshTcpRebalanceIntervalInMills());
- Assertions.assertEquals(18816, config.getEventMeshServerAdminPort());
Assertions.assertEquals(Boolean.TRUE,
config.isEventMeshTcpSendBackEnabled());
Assertions.assertEquals(3, config.getEventMeshTcpSendBackMaxTimes());
Assertions.assertEquals(21816,
config.getEventMeshTcpPushFailIsolateTimeInMills());
diff --git
a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/configuration/EventMeshGrpcConfigurationTest.java
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/configuration/EventMeshGrpcConfigurationTest.java
index 6b206167d..4dc59a200 100644
---
a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/configuration/EventMeshGrpcConfigurationTest.java
+++
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/configuration/EventMeshGrpcConfigurationTest.java
@@ -50,7 +50,6 @@ public class EventMeshGrpcConfigurationTest {
Assertions.assertEquals(5816,
config.getEventMeshServerReplyMsgThreadNum());
Assertions.assertEquals(6816,
config.getEventMeshServerSubscribeMsgThreadNum());
Assertions.assertEquals(7816,
config.getEventMeshServerMetaStorageThreadNum());
- Assertions.assertEquals(8816,
config.getEventMeshServerAdminThreadNum());
Assertions.assertEquals(9816,
config.getEventMeshServerRetryThreadNum());
Assertions.assertEquals(11816,
config.getEventMeshServerPullMetaStorageInterval());
Assertions.assertEquals(12816,
config.getEventMeshServerAsyncAccumulationThreshold());
diff --git
a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/configuration/EventMeshHTTPConfigurationTest.java
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/configuration/EventMeshHTTPConfigurationTest.java
index d522ff551..9e99bc511 100644
---
a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/configuration/EventMeshHTTPConfigurationTest.java
+++
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/configuration/EventMeshHTTPConfigurationTest.java
@@ -53,7 +53,6 @@ public class EventMeshHTTPConfigurationTest {
Assertions.assertEquals(5816,
config.getEventMeshServerReplyMsgThreadNum());
Assertions.assertEquals(6816,
config.getEventMeshServerClientManageThreadNum());
Assertions.assertEquals(7816,
config.getEventMeshServerMetaStorageThreadNum());
- Assertions.assertEquals(8816,
config.getEventMeshServerAdminThreadNum());
Assertions.assertEquals(9816,
config.getEventMeshServerRetryThreadNum());
Assertions.assertEquals(11816,
config.getEventMeshServerPullMetaStorageInterval());
diff --git
a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/configuration/EventMeshTCPConfigurationTest.java
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/configuration/EventMeshTCPConfigurationTest.java
index 1501cf1b5..2aee84ef7 100644
---
a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/configuration/EventMeshTCPConfigurationTest.java
+++
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/configuration/EventMeshTCPConfigurationTest.java
@@ -58,7 +58,6 @@ public class EventMeshTCPConfigurationTest {
Assertions.assertEquals(15816,
config.getEventMeshTcpMsgRetryAsyncDelayInMills());
Assertions.assertEquals(16816,
config.getEventMeshTcpMsgRetryQueueSize());
Assertions.assertEquals(Integer.valueOf(17816),
config.getEventMeshTcpRebalanceIntervalInMills());
- Assertions.assertEquals(18816, config.getEventMeshServerAdminPort());
Assertions.assertEquals(Boolean.TRUE,
config.isEventMeshTcpSendBackEnabled());
Assertions.assertEquals(3, config.getEventMeshTcpSendBackMaxTimes());
Assertions.assertEquals(21816,
config.getEventMeshTcpPushFailIsolateTimeInMills());
diff --git a/eventmesh-runtime/src/test/resources/configuration.properties
b/eventmesh-runtime/src/test/resources/configuration.properties
index 70ff82e05..836fc9c98 100644
--- a/eventmesh-runtime/src/test/resources/configuration.properties
+++ b/eventmesh-runtime/src/test/resources/configuration.properties
@@ -78,7 +78,6 @@ eventMesh.server.tcp.writerIdleSeconds=2816
eventMesh.server.tcp.readerIdleSeconds=3816
eventMesh.server.tcp.msgReqnumPerSecond=4816
eventMesh.server.tcp.clientMaxNum=5816
-eventMesh.server.tcp.enabled=true
eventMesh.server.global.scheduler=6816
eventMesh.server.tcp.taskHandleExecutorPoolSize=7816
eventMesh.server.tcp.msgDownStreamExecutorPoolSize=8816
@@ -89,7 +88,6 @@ eventMesh.server.retry.sync.pushRetryDelayInMills=14816
eventMesh.server.retry.async.pushRetryDelayInMills=15816
eventMesh.server.retry.pushRetryQueueSize=16816
eventMesh.server.tcp.RebalanceIntervalInMills=17816
-eventMesh.server.admin.http.port=18816
eventMesh.server.tcp.sendBack.enabled=true
eventMesh.server.tcp.pushFailIsolateTimeInMills=21816
eventMesh.server.gracefulShutdown.sleepIntervalInMills=22816
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
