[
https://issues.apache.org/jira/browse/SCB-1023?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16698724#comment-16698724
]
ASF GitHub Bot commented on SCB-1023:
-------------------------------------
wujimin closed pull request #1008: [SCB-1023] Max connection limit
URL: https://github.com/apache/servicecomb-java-chassis/pull/1008
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/core/src/main/java/org/apache/servicecomb/core/Transport.java
b/core/src/main/java/org/apache/servicecomb/core/Transport.java
index 026ed7c4e..ed9ade43e 100644
--- a/core/src/main/java/org/apache/servicecomb/core/Transport.java
+++ b/core/src/main/java/org/apache/servicecomb/core/Transport.java
@@ -17,8 +17,6 @@
package org.apache.servicecomb.core;
-import java.util.concurrent.atomic.AtomicInteger;
-
import org.apache.servicecomb.swagger.invocation.AsyncResponse;
// TODO:感觉要拆成显式的client、server才好些
@@ -51,6 +49,4 @@ default boolean canInit() {
Endpoint getPublishEndpoint() throws Exception;
void send(Invocation invocation, AsyncResponse asyncResp) throws Exception;
-
- AtomicInteger getConnectedCounter();
}
diff --git
a/core/src/main/java/org/apache/servicecomb/core/transport/AbstractTransport.java
b/core/src/main/java/org/apache/servicecomb/core/transport/AbstractTransport.java
index 2d6ab9872..3dc525c09 100644
---
a/core/src/main/java/org/apache/servicecomb/core/transport/AbstractTransport.java
+++
b/core/src/main/java/org/apache/servicecomb/core/transport/AbstractTransport.java
@@ -23,7 +23,6 @@
import java.net.URLDecoder;
import java.nio.charset.StandardCharsets;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import org.apache.http.client.utils.URLEncodedUtils;
@@ -71,8 +70,6 @@ public static TransportVertxFactory
getTransportVertxFactory() {
protected Endpoint publishEndpoint;
- private final AtomicInteger connectedCounter = new AtomicInteger(0);
-
@Override
public Endpoint getPublishEndpoint() {
return publishEndpoint;
@@ -83,11 +80,6 @@ public Endpoint getEndpoint() {
return endpoint;
}
- @Override
- public AtomicInteger getConnectedCounter() {
- return connectedCounter;
- }
-
protected void setListenAddressWithoutSchema(String addressWithoutSchema) {
setListenAddressWithoutSchema(addressWithoutSchema, null);
}
diff --git a/core/src/test/java/org/apache/servicecomb/core/TestTransport.java
b/core/src/test/java/org/apache/servicecomb/core/TestTransport.java
index abfa20252..d6a118308 100644
--- a/core/src/test/java/org/apache/servicecomb/core/TestTransport.java
+++ b/core/src/test/java/org/apache/servicecomb/core/TestTransport.java
@@ -21,7 +21,6 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
import org.apache.servicecomb.core.endpoint.EndpointsCache;
import org.apache.servicecomb.core.transport.TransportManager;
@@ -47,11 +46,6 @@ public void testEndpoint() throws Exception {
public void send(Invocation invocation, AsyncResponse asyncResp) {
}
- @Override
- public AtomicInteger getConnectedCounter() {
- return new AtomicInteger(0);
- }
-
@Override
public Object parseAddress(String address) {
return "127.0.0.1";
diff --git
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/ClientEvent.java
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/ClientEvent.java
deleted file mode 100644
index 0b37bbfcc..000000000
---
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/ClientEvent.java
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
- * 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.servicecomb.foundation.vertx;
-
-/**
- * Notice: this event will raised in vertx eventloop thread, so do not run any
block code
- */
-public class ClientEvent {
- private final String address;
-
- private final ConnectionEvent connectionEvent;
-
- private final TransportType transportType;
-
- private final int totalConnectedCount;
-
- public String getAddress() {
- return address;
- }
-
- public ConnectionEvent getConnectionEvent() {
- return connectionEvent;
- }
-
- public TransportType getTransportType() {
- return transportType;
- }
-
- public int getTotalConnectedCount() {
- return totalConnectedCount;
- }
-
- public ClientEvent(String address, ConnectionEvent connectionEvent,
TransportType transportType,
- int totalConnectedCount) {
- this.address = address;
- this.connectionEvent = connectionEvent;
- this.transportType = transportType;
- this.totalConnectedCount = totalConnectedCount;
- }
-}
\ No newline at end of file
diff --git
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/DefaultHttpServerMetrics.java
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/DefaultHttpServerMetrics.java
index 76692dd15..b99e8f924 100644
---
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/DefaultHttpServerMetrics.java
+++
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/DefaultHttpServerMetrics.java
@@ -16,8 +16,8 @@
*/
package org.apache.servicecomb.foundation.vertx.metrics;
-import
org.apache.servicecomb.foundation.vertx.metrics.metric.DefaultEndpointMetric;
import
org.apache.servicecomb.foundation.vertx.metrics.metric.DefaultHttpSocketMetric;
+import
org.apache.servicecomb.foundation.vertx.metrics.metric.DefaultServerEndpointMetric;
import io.vertx.core.http.HttpMethod;
import io.vertx.core.http.HttpServerRequest;
@@ -30,13 +30,13 @@
* important: not singleton, every HttpServer instance relate to a
HttpServerMetrics instance
*/
public class DefaultHttpServerMetrics implements HttpServerMetrics<Object,
Object, DefaultHttpSocketMetric> {
- private final DefaultEndpointMetric endpointMetric;
+ private final DefaultServerEndpointMetric endpointMetric;
- public DefaultHttpServerMetrics(DefaultEndpointMetric endpointMetric) {
+ public DefaultHttpServerMetrics(DefaultServerEndpointMetric endpointMetric) {
this.endpointMetric = endpointMetric;
}
- public DefaultEndpointMetric getEndpointMetric() {
+ public DefaultServerEndpointMetric getEndpointMetric() {
return endpointMetric;
}
diff --git
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/DefaultTcpServerMetrics.java
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/DefaultTcpServerMetrics.java
index d208c6d9f..a68c56f74 100644
---
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/DefaultTcpServerMetrics.java
+++
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/DefaultTcpServerMetrics.java
@@ -16,7 +16,7 @@
*/
package org.apache.servicecomb.foundation.vertx.metrics;
-import
org.apache.servicecomb.foundation.vertx.metrics.metric.DefaultEndpointMetric;
+import
org.apache.servicecomb.foundation.vertx.metrics.metric.DefaultServerEndpointMetric;
import
org.apache.servicecomb.foundation.vertx.metrics.metric.DefaultTcpSocketMetric;
import io.vertx.core.net.SocketAddress;
@@ -26,13 +26,13 @@
* important: not singleton, every NetServer instance relate to a
TcpServerMetrics instance
*/
public class DefaultTcpServerMetrics implements
TCPMetrics<DefaultTcpSocketMetric> {
- private final DefaultEndpointMetric endpointMetric;
+ private final DefaultServerEndpointMetric endpointMetric;
- public DefaultTcpServerMetrics(DefaultEndpointMetric endpointMetric) {
+ public DefaultTcpServerMetrics(DefaultServerEndpointMetric endpointMetric) {
this.endpointMetric = endpointMetric;
}
- public DefaultEndpointMetric getEndpointMetric() {
+ public DefaultServerEndpointMetric getEndpointMetric() {
return endpointMetric;
}
diff --git
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/DefaultVertxMetrics.java
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/DefaultVertxMetrics.java
index 41cec77c8..4d59a296f 100644
---
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/DefaultVertxMetrics.java
+++
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/DefaultVertxMetrics.java
@@ -20,7 +20,7 @@
import org.apache.servicecomb.foundation.common.concurrent.ConcurrentHashMapEx;
import
org.apache.servicecomb.foundation.vertx.metrics.metric.DefaultClientEndpointMetricManager;
-import
org.apache.servicecomb.foundation.vertx.metrics.metric.DefaultEndpointMetric;
+import
org.apache.servicecomb.foundation.vertx.metrics.metric.DefaultServerEndpointMetric;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
@@ -42,7 +42,7 @@
private VertxOptions vertxOptions;
// to support listen multiple addresses, must use a map to manage the metric
- private Map<SocketAddress, DefaultEndpointMetric> serverEndpointMetricMap =
new ConcurrentHashMapEx<>();
+ private Map<SocketAddress, DefaultServerEndpointMetric>
serverEndpointMetricMap = new ConcurrentHashMapEx<>();
private volatile DefaultClientEndpointMetricManager
clientEndpointMetricManager;
@@ -61,15 +61,15 @@ public DefaultClientEndpointMetricManager
getClientEndpointMetricManager() {
return clientEndpointMetricManager;
}
- public Map<SocketAddress, DefaultEndpointMetric>
getServerEndpointMetricMap() {
+ public Map<SocketAddress, DefaultServerEndpointMetric>
getServerEndpointMetricMap() {
return serverEndpointMetricMap;
}
@Override
public HttpServerMetrics<?, ?, ?> createMetrics(HttpServer server,
SocketAddress localAddress,
HttpServerOptions options) {
- DefaultEndpointMetric endpointMetric = serverEndpointMetricMap
- .computeIfAbsent(localAddress, DefaultEndpointMetric::new);
+ DefaultServerEndpointMetric endpointMetric = serverEndpointMetricMap
+ .computeIfAbsent(localAddress, DefaultServerEndpointMetric::new);
return new DefaultHttpServerMetrics(endpointMetric);
}
@@ -80,8 +80,8 @@ public DefaultClientEndpointMetricManager
getClientEndpointMetricManager() {
@Override
public TCPMetrics<?> createMetrics(SocketAddress localAddress,
NetServerOptions options) {
- DefaultEndpointMetric endpointMetric = serverEndpointMetricMap
- .computeIfAbsent(localAddress, DefaultEndpointMetric::new);
+ DefaultServerEndpointMetric endpointMetric = serverEndpointMetricMap
+ .computeIfAbsent(localAddress, DefaultServerEndpointMetric::new);
return new DefaultTcpServerMetrics(endpointMetric);
}
diff --git
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/MetricsOptionsEx.java
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/MetricsOptionsEx.java
index 702ca64d8..90da2c140 100644
---
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/MetricsOptionsEx.java
+++
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/MetricsOptionsEx.java
@@ -23,7 +23,7 @@
public class MetricsOptionsEx extends MetricsOptions {
private long checkClientEndpointMetricIntervalInMilliseconds =
TimeUnit.MINUTES.toMillis(1);
- private long checkClientEndpointMetricExpiredInNano =
TimeUnit.MINUTES.toNanos(30);
+ private long checkClientEndpointMetricExpiredInNano =
TimeUnit.MINUTES.toNanos(15);
public long getCheckClientEndpointMetricIntervalInMilliseconds() {
return checkClientEndpointMetricIntervalInMilliseconds;
diff --git
a/integration-tests/spring-pojo-tests/src/test/java/org/apache/servicecomb/demo/pojo/test/ConnectionEventWatcher.java
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/metric/DefaultServerEndpointMetric.java
similarity index 57%
rename from
integration-tests/spring-pojo-tests/src/test/java/org/apache/servicecomb/demo/pojo/test/ConnectionEventWatcher.java
rename to
foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/metric/DefaultServerEndpointMetric.java
index bda03c01c..b2d48c2d8 100644
---
a/integration-tests/spring-pojo-tests/src/test/java/org/apache/servicecomb/demo/pojo/test/ConnectionEventWatcher.java
+++
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/metrics/metric/DefaultServerEndpointMetric.java
@@ -14,28 +14,24 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.servicecomb.foundation.vertx.metrics.metric;
-package org.apache.servicecomb.demo.pojo.test;
+import java.util.concurrent.atomic.LongAdder;
-import java.util.ArrayList;
-import java.util.List;
+import io.vertx.core.net.SocketAddress;
-import org.apache.servicecomb.foundation.vertx.ClientEvent;
-import org.apache.servicecomb.foundation.vertx.TransportType;
+public class DefaultServerEndpointMetric extends DefaultEndpointMetric {
+ private LongAdder rejectByConnectionLimitCount = new LongAdder();
-import com.google.common.eventbus.Subscribe;
-
-public class ConnectionEventWatcher {
- private final List<Integer> counters = new ArrayList<>();
+ public DefaultServerEndpointMetric(SocketAddress address) {
+ super(address);
+ }
- public List<Integer> getCounters() {
- return counters;
+ public long getRejectByConnectionLimitCount() {
+ return rejectByConnectionLimitCount.longValue();
}
- @Subscribe
- public void getEvent(ClientEvent event) {
- if (TransportType.Highway.equals(event.getTransportType())) {
- counters.add(event.getTotalConnectedCount());
- }
+ public void onRejectByConnectionLimit() {
+ rejectByConnectionLimitCount.increment();
}
}
diff --git
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/server/TcpServer.java
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/server/TcpServer.java
index be0ff460e..58b36c943 100644
---
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/server/TcpServer.java
+++
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/server/TcpServer.java
@@ -18,38 +18,31 @@
package org.apache.servicecomb.foundation.vertx.server;
import java.net.InetSocketAddress;
-import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.servicecomb.foundation.common.event.EventManager;
import org.apache.servicecomb.foundation.common.net.URIEndpointObject;
import org.apache.servicecomb.foundation.common.utils.ExceptionUtils;
import org.apache.servicecomb.foundation.ssl.SSLCustom;
import org.apache.servicecomb.foundation.ssl.SSLOption;
import org.apache.servicecomb.foundation.ssl.SSLOptionFactory;
import org.apache.servicecomb.foundation.vertx.AsyncResultCallback;
-import org.apache.servicecomb.foundation.vertx.ClientEvent;
-import org.apache.servicecomb.foundation.vertx.ConnectionEvent;
-import org.apache.servicecomb.foundation.vertx.TransportType;
import org.apache.servicecomb.foundation.vertx.VertxTLSBuilder;
+import org.apache.servicecomb.foundation.vertx.metrics.DefaultTcpServerMetrics;
+import
org.apache.servicecomb.foundation.vertx.metrics.metric.DefaultServerEndpointMetric;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.netflix.config.DynamicPropertyFactory;
-
import io.vertx.core.Vertx;
import io.vertx.core.net.NetServer;
import io.vertx.core.net.NetServerOptions;
+import io.vertx.core.net.impl.NetSocketImpl;
public class TcpServer {
private static final Logger LOGGER =
LoggerFactory.getLogger(TcpServer.class);
private URIEndpointObject endpointObject;
- private final AtomicInteger connectedCounter;
-
- public TcpServer(URIEndpointObject endpointObject, AtomicInteger
connectedCounter) {
+ public TcpServer(URIEndpointObject endpointObject) {
this.endpointObject = endpointObject;
- this.connectedCounter = connectedCounter;
}
public void init(Vertx vertx, String sslKey,
AsyncResultCallback<InetSocketAddress> callback) {
@@ -72,19 +65,18 @@ public void init(Vertx vertx, String sslKey,
AsyncResultCallback<InetSocketAddre
}
netServer.connectHandler(netSocket -> {
- int connectedCount = connectedCounter.incrementAndGet();
- int connectionLimit = DynamicPropertyFactory.getInstance()
- .getIntProperty("servicecomb.highway.server.connection-limit",
Integer.MAX_VALUE).get();
+ DefaultTcpServerMetrics serverMetrics = (DefaultTcpServerMetrics)
((NetSocketImpl) netSocket).metrics();
+ DefaultServerEndpointMetric endpointMetric =
serverMetrics.getEndpointMetric();
+ long connectedCount = endpointMetric.getCurrentConnectionCount();
+ int connectionLimit = getConnectionLimit();
if (connectedCount > connectionLimit) {
- connectedCounter.decrementAndGet();
netSocket.close();
+ endpointMetric.onRejectByConnectionLimit();
return;
}
TcpServerConnection connection = createTcpServerConnection();
- connection.init(netSocket, connectedCounter);
- EventManager.post(new ClientEvent(netSocket.remoteAddress().toString(),
- ConnectionEvent.Connected, TransportType.Highway, connectedCount));
+ connection.init(netSocket);
});
netServer.exceptionHandler(e -> {
LOGGER.error("Unexpected error in server.{}",
ExceptionUtils.getExceptionMessageWithoutTrace(e));
@@ -102,6 +94,10 @@ public void init(Vertx vertx, String sslKey,
AsyncResultCallback<InetSocketAddre
});
}
+ protected int getConnectionLimit() {
+ return Integer.MAX_VALUE;
+ }
+
protected TcpServerConnection createTcpServerConnection() {
return new TcpServerConnection();
}
diff --git
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/server/TcpServerConnection.java
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/server/TcpServerConnection.java
index 18120a477..348a3f843 100644
---
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/server/TcpServerConnection.java
+++
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/server/TcpServerConnection.java
@@ -16,12 +16,6 @@
*/
package org.apache.servicecomb.foundation.vertx.server;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.apache.servicecomb.foundation.common.event.EventManager;
-import org.apache.servicecomb.foundation.vertx.ClientEvent;
-import org.apache.servicecomb.foundation.vertx.ConnectionEvent;
-import org.apache.servicecomb.foundation.vertx.TransportType;
import org.apache.servicecomb.foundation.vertx.tcp.TcpConnection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -34,7 +28,7 @@
protected TcpParser splitter;
- public void init(NetSocket netSocket, AtomicInteger connectedCounter) {
+ public void init(NetSocket netSocket) {
// currently, socket always be NetSocketImpl
this.initNetSocket((NetSocketImpl) netSocket);
@@ -52,9 +46,6 @@ public void init(NetSocket netSocket, AtomicInteger
connectedCounter) {
LOGGER.error("disconected from {}, in thread {}",
remoteAddress,
Thread.currentThread().getName());
-
- int connectedCount = connectedCounter.decrementAndGet();
- EventManager.post(new ClientEvent(remoteAddress, ConnectionEvent.Closed,
TransportType.Highway, connectedCount));
});
netSocket.handler(splitter);
diff --git
a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/metrics/TestMetricsOptionsEx.java
b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/metrics/TestMetricsOptionsEx.java
index cced3533b..19f3241ac 100644
---
a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/metrics/TestMetricsOptionsEx.java
+++
b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/metrics/TestMetricsOptionsEx.java
@@ -36,7 +36,7 @@ public void interval() {
@Test
public void expired() {
- Assert.assertEquals(TimeUnit.MINUTES.toNanos(30),
metricsOptionsEx.getCheckClientEndpointMetricExpiredInNano());
+ Assert.assertEquals(TimeUnit.MINUTES.toNanos(15),
metricsOptionsEx.getCheckClientEndpointMetricExpiredInNano());
metricsOptionsEx.setCheckClientEndpointMetricExpiredInNano(10);
Assert.assertEquals(10,
metricsOptionsEx.getCheckClientEndpointMetricExpiredInNano());
diff --git
a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/server/TestTcpServer.java
b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/server/TestTcpServer.java
index 59825f5c9..b95bfc30a 100644
---
a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/server/TestTcpServer.java
+++
b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/server/TestTcpServer.java
@@ -18,10 +18,12 @@
package org.apache.servicecomb.foundation.vertx.server;
import java.net.InetSocketAddress;
-import java.util.concurrent.atomic.AtomicInteger;
import org.apache.servicecomb.foundation.common.net.URIEndpointObject;
import org.apache.servicecomb.foundation.vertx.AsyncResultCallback;
+import org.apache.servicecomb.foundation.vertx.metrics.DefaultTcpServerMetrics;
+import
org.apache.servicecomb.foundation.vertx.metrics.metric.DefaultServerEndpointMetric;
+import org.junit.Assert;
import org.junit.Test;
import io.vertx.core.Handler;
@@ -29,21 +31,24 @@
import io.vertx.core.net.NetServer;
import io.vertx.core.net.NetServerOptions;
import io.vertx.core.net.NetSocket;
+import io.vertx.core.net.impl.NetSocketImpl;
import mockit.Expectations;
+import mockit.Mock;
+import mockit.MockUp;
import mockit.Mocked;
public class TestTcpServer {
static class TcpServerForTest extends TcpServer {
public TcpServerForTest(URIEndpointObject endpointObject) {
- super(endpointObject, new AtomicInteger());
+ super(endpointObject);
}
@Override
protected TcpServerConnection createTcpServerConnection() {
return new TcpServerConnection() {
@Override
- public void init(NetSocket netSocket, AtomicInteger connectedCounter) {
- super.init(netSocket, connectedCounter);
+ public void init(NetSocket netSocket) {
+ super.init(netSocket);
}
};
}
@@ -84,4 +89,59 @@ public void testTcpServerSSL(@Mocked Vertx vertx, @Mocked
AsyncResultCallback<In
// assert done in Expectations
server.init(vertx, "", callback);
}
+
+ Handler<NetSocket> connectHandler;
+
+ boolean netSocketClosed;
+
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ @Test
+ public void testConnectionLimit(@Mocked Vertx vertx, @Mocked
AsyncResultCallback<InetSocketAddress> callback,
+ @Mocked NetServer netServer, @Mocked NetSocketImpl netSocket) {
+ DefaultServerEndpointMetric endpointMetric = new
DefaultServerEndpointMetric(null);
+ DefaultTcpServerMetrics tcpServerMetrics = new
DefaultTcpServerMetrics(endpointMetric);
+
+ new MockUp<NetServer>(netServer) {
+ @Mock
+ NetServer connectHandler(Handler<NetSocket> handler) {
+ connectHandler = handler;
+ return netServer;
+ }
+ };
+ new MockUp<NetSocketImpl>(netSocket) {
+ @Mock
+ void close() {
+ netSocketClosed = true;
+ }
+ };
+ new Expectations() {
+ {
+ vertx.createNetServer((NetServerOptions) any);
+ result = netServer;
+ netServer.listen(anyInt, anyString, (Handler) any);
+ netSocket.metrics();
+ result = tcpServerMetrics;
+ }
+ };
+ URIEndpointObject endpointObject = new
URIEndpointObject("highway://127.0.0.1:6663?sslEnabled=true");
+ TcpServer server = new TcpServerForTest(endpointObject) {
+ @Override
+ protected int getConnectionLimit() {
+ return 2;
+ }
+ };
+ // assert done in Expectations
+ server.init(vertx, "", callback);
+
+ // no problem
+ endpointMetric.onConnect();
+ endpointMetric.onConnect();
+ connectHandler.handle(netSocket);
+
+ // reject
+ endpointMetric.onConnect();
+ connectHandler.handle(netSocket);
+ Assert.assertTrue(netSocketClosed);
+ Assert.assertEquals(1, endpointMetric.getRejectByConnectionLimitCount());
+ }
}
diff --git
a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/server/TestTcpServerConnection.java
b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/server/TestTcpServerConnection.java
index 08b39780b..3047042f2 100644
---
a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/server/TestTcpServerConnection.java
+++
b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/server/TestTcpServerConnection.java
@@ -16,8 +16,6 @@
*/
package org.apache.servicecomb.foundation.vertx.server;
-import java.util.concurrent.atomic.AtomicInteger;
-
import org.junit.Assert;
import org.junit.Test;
@@ -31,7 +29,7 @@ public void test(@Mocked NetSocketImpl netSocket) {
connection.setProtocol("p");
connection.setZipName("z");
- connection.init(netSocket, new AtomicInteger());
+ connection.init(netSocket);
Assert.assertEquals(netSocket, connection.getNetSocket());
}
diff --git
a/integration-tests/spring-pojo-tests/src/test/java/org/apache/servicecomb/demo/pojo/test/PojoSpringIntegrationTest.java
b/integration-tests/spring-pojo-tests/src/test/java/org/apache/servicecomb/demo/pojo/test/PojoSpringIntegrationTest.java
index bcd56e73a..388ed6867 100644
---
a/integration-tests/spring-pojo-tests/src/test/java/org/apache/servicecomb/demo/pojo/test/PojoSpringIntegrationTest.java
+++
b/integration-tests/spring-pojo-tests/src/test/java/org/apache/servicecomb/demo/pojo/test/PojoSpringIntegrationTest.java
@@ -18,9 +18,7 @@
package org.apache.servicecomb.demo.pojo.test;
import org.apache.servicecomb.core.SCBEngine;
-import org.apache.servicecomb.foundation.common.event.EventManager;
import org.junit.AfterClass;
-import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.runner.RunWith;
import org.springframework.test.context.junit4.SpringRunner;
@@ -28,18 +26,14 @@
@RunWith(SpringRunner.class)
public class PojoSpringIntegrationTest extends PojoIntegrationTestBase {
- private static final ConnectionEventWatcher watcher = new
ConnectionEventWatcher();
-
@BeforeClass
public static void setUpClass() throws Exception {
setUpLocalRegistry();
- EventManager.register(watcher);
PojoTestMain.main(null);
}
@AfterClass
public static void teardownClass() {
SCBEngine.getInstance().destroy();
- Assert.assertArrayEquals("check connection count change", new Integer[]
{1, 0}, watcher.getCounters().toArray());
}
-}
+}
\ No newline at end of file
diff --git
a/integration-tests/springmvc-tests/springmvc-tests-general/src/test/java/org/apache/servicecomb/demo/springmvc/tests/RawSpringMvcIntegrationTest.java
b/integration-tests/springmvc-tests/springmvc-tests-general/src/test/java/org/apache/servicecomb/demo/springmvc/tests/RawSpringMvcIntegrationTest.java
index 14d81da68..30924d7f8 100644
---
a/integration-tests/springmvc-tests/springmvc-tests-general/src/test/java/org/apache/servicecomb/demo/springmvc/tests/RawSpringMvcIntegrationTest.java
+++
b/integration-tests/springmvc-tests/springmvc-tests-general/src/test/java/org/apache/servicecomb/demo/springmvc/tests/RawSpringMvcIntegrationTest.java
@@ -18,27 +18,19 @@
package org.apache.servicecomb.demo.springmvc.tests;
import org.apache.servicecomb.core.SCBEngine;
-import org.apache.servicecomb.foundation.common.event.EventManager;
import org.junit.AfterClass;
-import org.junit.Assert;
import org.junit.BeforeClass;
public class RawSpringMvcIntegrationTest extends SpringMvcIntegrationTestBase {
-
- private static final ConnectionEventWatcher watcher = new
ConnectionEventWatcher();
-
@BeforeClass
public static void init() throws Exception {
System.setProperty("servicecomb.uploads.directory", "/tmp");
setUpLocalRegistry();
- EventManager.register(watcher);
SpringMvcTestMain.main(new String[0]);
}
@AfterClass
public static void shutdown() {
SCBEngine.getInstance().destroy();
- Assert.assertArrayEquals("check connection count change", new Integer[]
{1, 2, 1, 0},
- watcher.getCounters().toArray());
}
}
diff --git
a/integration-tests/springmvc-tests/springmvc-tests-simplified-mapping/src/test/java/org/apache/servicecomb/demo/springmvc/tests/RawSpringMvcSimplifiedMappingAnnotationIntegrationTest.java
b/integration-tests/springmvc-tests/springmvc-tests-simplified-mapping/src/test/java/org/apache/servicecomb/demo/springmvc/tests/RawSpringMvcSimplifiedMappingAnnotationIntegrationTest.java
index 32cfb03e7..f167347fc 100644
---
a/integration-tests/springmvc-tests/springmvc-tests-simplified-mapping/src/test/java/org/apache/servicecomb/demo/springmvc/tests/RawSpringMvcSimplifiedMappingAnnotationIntegrationTest.java
+++
b/integration-tests/springmvc-tests/springmvc-tests-simplified-mapping/src/test/java/org/apache/servicecomb/demo/springmvc/tests/RawSpringMvcSimplifiedMappingAnnotationIntegrationTest.java
@@ -18,28 +18,20 @@
package org.apache.servicecomb.demo.springmvc.tests;
import org.apache.servicecomb.core.SCBEngine;
-import org.apache.servicecomb.foundation.common.event.EventManager;
import org.junit.AfterClass;
-import org.junit.Assert;
import org.junit.BeforeClass;
public class RawSpringMvcSimplifiedMappingAnnotationIntegrationTest extends
SpringMvcIntegrationTestBase {
-
- private static final ConnectionEventWatcher watcher = new
ConnectionEventWatcher();
-
@BeforeClass
public static void init() throws Exception {
System.setProperty("spring.profiles.active", "SimplifiedMapping");
System.setProperty("servicecomb.uploads.directory", "/tmp");
setUpLocalRegistry();
- EventManager.register(watcher);
SpringMvcTestMain.main(new String[0]);
}
@AfterClass
public static void shutdown() {
SCBEngine.getInstance().destroy();
- Assert.assertArrayEquals("check connection count change", new Integer[]
{1, 2, 1, 0},
- watcher.getCounters().toArray());
}
}
diff --git
a/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/VertxMetersInitializer.java
b/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/VertxMetersInitializer.java
index bd3c5676d..fc8b4df6f 100644
---
a/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/VertxMetersInitializer.java
+++
b/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/VertxMetersInitializer.java
@@ -21,6 +21,7 @@
import org.apache.servicecomb.foundation.metrics.MetricsInitializer;
import org.apache.servicecomb.foundation.metrics.registry.GlobalRegistry;
import org.apache.servicecomb.metrics.core.meter.vertx.VertxEndpointsMeter;
+import
org.apache.servicecomb.metrics.core.meter.vertx.VertxServerEndpointsMeter;
import com.google.common.eventbus.EventBus;
import com.netflix.spectator.api.Id;
@@ -50,7 +51,8 @@ public void init(GlobalRegistry globalRegistry, EventBus
eventBus, MetricsBootst
.getClientEndpointMetricMap());
SpectatorUtils.registerMeter(registry, clientMeter);
- VertxEndpointsMeter serverMeter = new
VertxEndpointsMeter(endpointsId.withTag(ENDPOINTS_TYPE, ENDPOINTS_SERVER),
+ VertxEndpointsMeter serverMeter = new VertxServerEndpointsMeter(
+ endpointsId.withTag(ENDPOINTS_TYPE, ENDPOINTS_SERVER),
AbstractTransport
.getTransportVertxFactory()
.getMetricsFactory()
diff --git
a/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/meter/vertx/EndpointMeter.java
b/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/meter/vertx/EndpointMeter.java
index e3e7cf571..8789a6107 100644
---
a/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/meter/vertx/EndpointMeter.java
+++
b/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/meter/vertx/EndpointMeter.java
@@ -38,6 +38,8 @@
public static final String BYTES_WRITTEN = "bytesWritten";
+ protected Id id;
+
private Id idConnect;
private Id idDisconnect;
@@ -48,7 +50,7 @@
private Id idBytesWritten;
- private DefaultEndpointMetric metric;
+ protected DefaultEndpointMetric metric;
private long lastConnectCount;
@@ -60,6 +62,7 @@
public EndpointMeter(Id id, DefaultEndpointMetric metric) {
id = id.withTag(ADDRESS, metric.getAddress().toString());
+ this.id = id;
idConnect = id.withTag(STATISTIC, CONNECT_COUNT);
idDisconnect = id.withTag(STATISTIC, DISCONNECT_COUNT);
idConnections = id.withTag(STATISTIC, CONNECTIONS);
@@ -72,7 +75,7 @@ public DefaultEndpointMetric getMetric() {
return metric;
}
- private Measurement newMeasurement(Id id, long timestamp, Number n) {
+ protected Measurement newMeasurement(Id id, long timestamp, Number n) {
return new Measurement(id, timestamp, n.doubleValue());
}
diff --git
a/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/meter/vertx/ServerEndpointMeter.java
b/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/meter/vertx/ServerEndpointMeter.java
new file mode 100644
index 000000000..300fe0398
--- /dev/null
+++
b/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/meter/vertx/ServerEndpointMeter.java
@@ -0,0 +1,48 @@
+/*
+ * 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.servicecomb.metrics.core.meter.vertx;
+
+import java.util.List;
+
+import
org.apache.servicecomb.foundation.vertx.metrics.metric.DefaultEndpointMetric;
+import
org.apache.servicecomb.foundation.vertx.metrics.metric.DefaultServerEndpointMetric;
+
+import com.netflix.spectator.api.Id;
+import com.netflix.spectator.api.Measurement;
+
+public class ServerEndpointMeter extends EndpointMeter {
+ public static final String REJECT_BY_CONNECTION_LIMIT =
"rejectByConnectionLimit";
+
+ private Id idRejectByConnectionLimit;
+
+ private long lastRejectByConnectionLimit;
+
+ public ServerEndpointMeter(Id id, DefaultEndpointMetric metric) {
+ super(id, metric);
+ idRejectByConnectionLimit = this.id.withTag(STATISTIC,
REJECT_BY_CONNECTION_LIMIT);
+ }
+
+ @Override
+ public void calcMeasurements(List<Measurement> measurements, long msNow,
double secondInterval) {
+ super.calcMeasurements(measurements, msNow, secondInterval);
+
+ long rejectByConnectionLimit = ((DefaultServerEndpointMetric)
metric).getRejectByConnectionLimitCount();
+ measurements
+ .add(newMeasurement(idRejectByConnectionLimit, msNow,
rejectByConnectionLimit - lastRejectByConnectionLimit));
+ this.lastRejectByConnectionLimit = rejectByConnectionLimit;
+ }
+}
diff --git
a/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/meter/vertx/VertxEndpointsMeter.java
b/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/meter/vertx/VertxEndpointsMeter.java
index 44014f840..abf25d187 100644
---
a/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/meter/vertx/VertxEndpointsMeter.java
+++
b/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/meter/vertx/VertxEndpointsMeter.java
@@ -63,10 +63,14 @@ private void syncMeters() {
}
}
for (DefaultEndpointMetric metric : endpointMetricMap.values()) {
- endpointMeterMap.computeIfAbsent(metric.getAddress(), addr -> new
EndpointMeter(id, metric));
+ endpointMeterMap.computeIfAbsent(metric.getAddress(), addr ->
createEndpointMeter(id, metric));
}
}
+ protected EndpointMeter createEndpointMeter(Id id, DefaultEndpointMetric
metric) {
+ return new EndpointMeter(id, metric);
+ }
+
@Override
public Iterable<Measurement> measure() {
return allMeasurements;
diff --git
a/integration-tests/springmvc-tests/springmvc-tests-common/src/test/java/org/apache/servicecomb/demo/springmvc/tests/ConnectionEventWatcher.java
b/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/meter/vertx/VertxServerEndpointsMeter.java
similarity index 57%
rename from
integration-tests/springmvc-tests/springmvc-tests-common/src/test/java/org/apache/servicecomb/demo/springmvc/tests/ConnectionEventWatcher.java
rename to
metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/meter/vertx/VertxServerEndpointsMeter.java
index 2ad819c17..1def3b232 100644
---
a/integration-tests/springmvc-tests/springmvc-tests-common/src/test/java/org/apache/servicecomb/demo/springmvc/tests/ConnectionEventWatcher.java
+++
b/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/meter/vertx/VertxServerEndpointsMeter.java
@@ -14,28 +14,24 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.servicecomb.metrics.core.meter.vertx;
-package org.apache.servicecomb.demo.springmvc.tests;
+import java.util.Map;
-import java.util.ArrayList;
-import java.util.List;
+import
org.apache.servicecomb.foundation.vertx.metrics.metric.DefaultEndpointMetric;
-import org.apache.servicecomb.foundation.vertx.ClientEvent;
-import org.apache.servicecomb.foundation.vertx.TransportType;
+import com.netflix.spectator.api.Id;
-import com.google.common.eventbus.Subscribe;
+import io.vertx.core.net.SocketAddress;
-public class ConnectionEventWatcher {
- private final List<Integer> counters = new ArrayList<>();
-
- public List<Integer> getCounters() {
- return counters;
+public class VertxServerEndpointsMeter extends VertxEndpointsMeter {
+ public <T extends DefaultEndpointMetric> VertxServerEndpointsMeter(
+ Id id, Map<SocketAddress, T> endpointMetricMap) {
+ super(id, endpointMetricMap);
}
- @Subscribe
- public void getEvent(ClientEvent event) {
- if (TransportType.Rest.equals(event.getTransportType())) {
- counters.add(event.getTotalConnectedCount());
- }
+ @Override
+ protected EndpointMeter createEndpointMeter(Id id, DefaultEndpointMetric
metric) {
+ return new ServerEndpointMeter(id, metric);
}
}
diff --git
a/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/publish/EndpointsLogPublisher.java
b/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/publish/ClientEndpointsLogPublisher.java
similarity index 84%
rename from
metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/publish/EndpointsLogPublisher.java
rename to
metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/publish/ClientEndpointsLogPublisher.java
index 5fcf60174..7bd9a874c 100644
---
a/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/publish/EndpointsLogPublisher.java
+++
b/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/publish/ClientEndpointsLogPublisher.java
@@ -24,23 +24,15 @@
import org.apache.servicecomb.metrics.core.VertxMetersInitializer;
import org.apache.servicecomb.metrics.core.meter.vertx.EndpointMeter;
-public class EndpointsLogPublisher extends AbstractMeasurementNodeLogPublisher
{
- private final String title;
-
- private final String endpointTitle;
-
- public EndpointsLogPublisher(MeasurementTree tree, StringBuilder sb, String
meterName, String title,
- String endpointTitle) {
+public class ClientEndpointsLogPublisher extends
AbstractMeasurementNodeLogPublisher {
+ public ClientEndpointsLogPublisher(MeasurementTree tree, StringBuilder sb,
String meterName) {
super(tree, sb, VertxMetersInitializer.VERTX_ENDPOINTS, meterName);
- this.title = title;
- this.endpointTitle = endpointTitle;
}
@Override
public void print(boolean printDetail) {
- appendLine(sb, " %s:", title);
- appendLine(sb, " %-21s connectCount disconnectCount connections
send receive",
- endpointTitle);
+ appendLine(sb, " client.endpoints:");
+ appendLine(sb, " remote connectCount
disconnectCount connections send receive");
double connect = 0;
double disconnect = 0;
diff --git
a/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/publish/DefaultLogPublisher.java
b/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/publish/DefaultLogPublisher.java
index 75a66422d..778a8bb53 100644
---
a/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/publish/DefaultLogPublisher.java
+++
b/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/publish/DefaultLogPublisher.java
@@ -425,10 +425,10 @@ protected void printVertxMetrics(MeasurementTree tree,
StringBuilder sb) {
entry.getValue().getEventLoopContextCreatedCount());
}
- EndpointsLogPublisher client = new EndpointsLogPublisher(tree, sb,
VertxMetersInitializer.ENDPOINTS_CLINET,
- "client.endpoints", "remote");
- EndpointsLogPublisher server = new EndpointsLogPublisher(tree, sb,
VertxMetersInitializer.ENDPOINTS_SERVER,
- "server.endpoints", "listen");
+ ClientEndpointsLogPublisher client = new ClientEndpointsLogPublisher(tree,
sb,
+ VertxMetersInitializer.ENDPOINTS_CLINET);
+ ServerEndpointsLogPublisher server = new ServerEndpointsLogPublisher(tree,
sb,
+ VertxMetersInitializer.ENDPOINTS_SERVER);
if (client.isExists() || server.isExists()) {
appendLine(sb, " transport:");
if (client.isExists()) {
diff --git
a/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/publish/ServerEndpointsLogPublisher.java
b/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/publish/ServerEndpointsLogPublisher.java
new file mode 100644
index 000000000..8edb940b0
--- /dev/null
+++
b/metrics/metrics-core/src/main/java/org/apache/servicecomb/metrics/core/publish/ServerEndpointsLogPublisher.java
@@ -0,0 +1,76 @@
+/*
+ * 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.servicecomb.metrics.core.publish;
+
+import static
org.apache.servicecomb.foundation.common.utils.StringBuilderUtils.appendLine;
+
+import org.apache.servicecomb.foundation.common.net.NetUtils;
+import
org.apache.servicecomb.foundation.metrics.publish.spectator.MeasurementNode;
+import
org.apache.servicecomb.foundation.metrics.publish.spectator.MeasurementTree;
+import org.apache.servicecomb.metrics.core.VertxMetersInitializer;
+import org.apache.servicecomb.metrics.core.meter.vertx.EndpointMeter;
+import org.apache.servicecomb.metrics.core.meter.vertx.ServerEndpointMeter;
+
+public class ServerEndpointsLogPublisher extends
AbstractMeasurementNodeLogPublisher {
+ public ServerEndpointsLogPublisher(MeasurementTree tree, StringBuilder sb,
String meterName) {
+ super(tree, sb, VertxMetersInitializer.VERTX_ENDPOINTS, meterName);
+ }
+
+ @Override
+ public void print(boolean printDetail) {
+ appendLine(sb, " server.endpoints:");
+ appendLine(sb,
+ " listen connectCount disconnectCount
rejectByLimit connections send receive");
+
+ double connect = 0;
+ double disconnect = 0;
+ double reject = 0;
+ double connections = 0;
+ double readSize = 0;
+ double writeSize = 0;
+ for (MeasurementNode address : measurementNode.getChildren().values()) {
+ connect += address.findChild(EndpointMeter.CONNECT_COUNT).summary();
+ disconnect +=
address.findChild(EndpointMeter.DISCONNECT_COUNT).summary();
+ reject +=
address.findChild(ServerEndpointMeter.REJECT_BY_CONNECTION_LIMIT).summary();
+ connections += address.findChild(EndpointMeter.CONNECTIONS).summary();
+ readSize += address.findChild(EndpointMeter.BYTES_READ).summary();
+ writeSize += address.findChild(EndpointMeter.BYTES_WRITTEN).summary();
+
+ if (printDetail) {
+ appendLine(sb, " %-21s %-15.0f %-15.0f %-15.0f %-12.0f %-12s
%-12s",
+ address.getName(),
+ address.findChild(EndpointMeter.CONNECT_COUNT).summary(),
+ address.findChild(EndpointMeter.DISCONNECT_COUNT).summary(),
+
address.findChild(ServerEndpointMeter.REJECT_BY_CONNECTION_LIMIT).summary(),
+ address.findChild(EndpointMeter.CONNECTIONS).summary(),
+ NetUtils.humanReadableBytes((long)
address.findChild(EndpointMeter.BYTES_WRITTEN).summary()),
+ NetUtils.humanReadableBytes((long)
address.findChild(EndpointMeter.BYTES_READ).summary())
+ );
+ }
+ }
+
+ appendLine(sb, " %-21s %-15.0f %-15.0f %-15.0f %-12.0f %-12s %-12s",
+ "(summary)",
+ connect,
+ disconnect,
+ reject,
+ connections,
+ NetUtils.humanReadableBytes((long) writeSize),
+ NetUtils.humanReadableBytes((long) readSize)
+ );
+ }
+}
diff --git
a/metrics/metrics-core/src/test/java/org/apache/servicecomb/metrics/core/TestVertxMetersInitializer.java
b/metrics/metrics-core/src/test/java/org/apache/servicecomb/metrics/core/TestVertxMetersInitializer.java
index 16c316c33..246e8a40d 100644
---
a/metrics/metrics-core/src/test/java/org/apache/servicecomb/metrics/core/TestVertxMetersInitializer.java
+++
b/metrics/metrics-core/src/test/java/org/apache/servicecomb/metrics/core/TestVertxMetersInitializer.java
@@ -166,9 +166,9 @@ private void testLog(LogCollector logCollector, List<Meter>
meters, List<Measure
}
expect += " (summary) 1 0 1
4 B 21 B \n"
+ " server.endpoints:\n"
- + " listen connectCount disconnectCount
connections send receive\n"
- + " 0.0.0.0:0 1 0 1
21 B 4 B \n"
- + " (summary) 1 0 1
21 B 4 B \n\n";
+ + " listen connectCount disconnectCount
rejectByLimit connections send receive\n"
+ + " 0.0.0.0:0 1 0 0
1 21 B 4 B \n"
+ + " (summary) 1 0 0
1 21 B 4 B \n\n";
Assert.assertEquals(expect, actual);
}
}
diff --git
a/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayServer.java
b/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayServer.java
index 0694c2de3..57a5a09a1 100644
---
a/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayServer.java
+++
b/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayServer.java
@@ -17,18 +17,18 @@
package org.apache.servicecomb.transport.highway;
-import java.util.concurrent.atomic.AtomicInteger;
-
import org.apache.servicecomb.core.Endpoint;
import org.apache.servicecomb.foundation.common.net.URIEndpointObject;
import org.apache.servicecomb.foundation.vertx.server.TcpServer;
import org.apache.servicecomb.foundation.vertx.server.TcpServerConnection;
+import com.netflix.config.DynamicPropertyFactory;
+
public class HighwayServer extends TcpServer {
private Endpoint endpoint;
- public HighwayServer(Endpoint endpoint, AtomicInteger connectedCounter) {
- super((URIEndpointObject) endpoint.getAddress(), connectedCounter);
+ public HighwayServer(Endpoint endpoint) {
+ super((URIEndpointObject) endpoint.getAddress());
this.endpoint = endpoint;
}
@@ -36,4 +36,10 @@ public HighwayServer(Endpoint endpoint, AtomicInteger
connectedCounter) {
protected TcpServerConnection createTcpServerConnection() {
return new HighwayServerConnection(endpoint);
}
+
+ @Override
+ protected int getConnectionLimit() {
+ return DynamicPropertyFactory.getInstance()
+ .getIntProperty("servicecomb.highway.server.connection-limit",
Integer.MAX_VALUE).get();
+ }
}
diff --git
a/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayServerConnection.java
b/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayServerConnection.java
index 36b80fa5f..5641992a6 100644
---
a/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayServerConnection.java
+++
b/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayServerConnection.java
@@ -16,8 +16,6 @@
*/
package org.apache.servicecomb.transport.highway;
-import java.util.concurrent.atomic.AtomicInteger;
-
import javax.ws.rs.core.Response.Status;
import org.apache.servicecomb.core.Endpoint;
@@ -44,9 +42,9 @@ public HighwayServerConnection(Endpoint endpoint) {
}
@Override
- public void init(NetSocket netSocket, AtomicInteger connectedCounter) {
+ public void init(NetSocket netSocket) {
splitter = new TcpParser(this);
- super.init(netSocket, connectedCounter);
+ super.init(netSocket);
}
@Override
diff --git
a/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayServerVerticle.java
b/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayServerVerticle.java
index 9a17081d5..3d6efc4cf 100644
---
a/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayServerVerticle.java
+++
b/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayServerVerticle.java
@@ -18,10 +18,8 @@
package org.apache.servicecomb.transport.highway;
import java.net.InetSocketAddress;
-import java.util.concurrent.atomic.AtomicInteger;
import org.apache.servicecomb.core.Const;
-import org.apache.servicecomb.core.CseContext;
import org.apache.servicecomb.core.Endpoint;
import org.apache.servicecomb.core.transport.AbstractTransport;
import org.apache.servicecomb.foundation.common.net.URIEndpointObject;
@@ -42,16 +40,6 @@
private URIEndpointObject endpointObject;
- private final AtomicInteger connectedCounter;
-
- public HighwayServerVerticle() {
-
this(CseContext.getInstance().getTransportManager().findTransport(Const.HIGHWAY).getConnectedCounter());
- }
-
- public HighwayServerVerticle(AtomicInteger connectedCounter) {
- this.connectedCounter = connectedCounter;
- }
-
@Override
public void init(Vertx vertx, Context context) {
super.init(vertx, context);
@@ -79,7 +67,7 @@ protected void startListen(Future<Void> startFuture) {
return;
}
- HighwayServer server = new HighwayServer(endpoint, connectedCounter);
+ HighwayServer server = new HighwayServer(endpoint);
server.init(vertx, SSL_KEY, ar -> {
if (ar.succeeded()) {
InetSocketAddress socketAddress = ar.result();
diff --git
a/transports/transport-highway/src/test/java/org/apache/servicecomb/transport/highway/TestHighwayServerConnection.java
b/transports/transport-highway/src/test/java/org/apache/servicecomb/transport/highway/TestHighwayServerConnection.java
index 9a18f9a7f..5375b0a46 100644
---
a/transports/transport-highway/src/test/java/org/apache/servicecomb/transport/highway/TestHighwayServerConnection.java
+++
b/transports/transport-highway/src/test/java/org/apache/servicecomb/transport/highway/TestHighwayServerConnection.java
@@ -17,7 +17,6 @@
package org.apache.servicecomb.transport.highway;
import java.net.InetSocketAddress;
-import java.util.concurrent.atomic.AtomicInteger;
import javax.xml.ws.Holder;
@@ -73,7 +72,7 @@ public void init() {
}
};
connection = new HighwayServerConnection(endpoint);
- connection.init(netSocket, new AtomicInteger());
+ connection.init(netSocket);
header = new RequestHeader();
}
diff --git
a/transports/transport-highway/src/test/java/org/apache/servicecomb/transport/highway/TestHighwayVerticle.java
b/transports/transport-highway/src/test/java/org/apache/servicecomb/transport/highway/TestHighwayVerticle.java
index 95019d8cc..f2ac8d6a1 100644
---
a/transports/transport-highway/src/test/java/org/apache/servicecomb/transport/highway/TestHighwayVerticle.java
+++
b/transports/transport-highway/src/test/java/org/apache/servicecomb/transport/highway/TestHighwayVerticle.java
@@ -19,8 +19,6 @@
import static org.junit.Assert.assertTrue;
-import java.util.concurrent.atomic.AtomicInteger;
-
import org.apache.servicecomb.core.Endpoint;
import org.apache.servicecomb.core.Transport;
import org.apache.servicecomb.foundation.common.net.URIEndpointObject;
@@ -40,7 +38,7 @@
@Test
public void testHighwayVerticle(@Mocked Transport transport, @Mocked Vertx
vertx, @Mocked Context context,
@Mocked JsonObject json) {
- HighwayServerVerticle highwayVerticle = new HighwayServerVerticle(new
AtomicInteger());
+ HighwayServerVerticle highwayVerticle = new HighwayServerVerticle();
URIEndpointObject endpiontObject = new
URIEndpointObject("highway://127.0.0.1:9090");
new Expectations() {
{
diff --git
a/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/RestServerVerticle.java
b/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/RestServerVerticle.java
index a61577d8c..a17f2874a 100644
---
a/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/RestServerVerticle.java
+++
b/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/RestServerVerticle.java
@@ -19,26 +19,21 @@
import java.util.List;
import java.util.Set;
-import java.util.concurrent.atomic.AtomicInteger;
import javax.ws.rs.core.MediaType;
import org.apache.servicecomb.common.rest.codec.RestObjectMapperFactory;
-import org.apache.servicecomb.core.Const;
-import org.apache.servicecomb.core.CseContext;
import org.apache.servicecomb.core.Endpoint;
import org.apache.servicecomb.core.transport.AbstractTransport;
-import org.apache.servicecomb.foundation.common.event.EventManager;
import org.apache.servicecomb.foundation.common.net.URIEndpointObject;
import org.apache.servicecomb.foundation.common.utils.ExceptionUtils;
import org.apache.servicecomb.foundation.common.utils.SPIServiceUtils;
import org.apache.servicecomb.foundation.ssl.SSLCustom;
import org.apache.servicecomb.foundation.ssl.SSLOption;
import org.apache.servicecomb.foundation.ssl.SSLOptionFactory;
-import org.apache.servicecomb.foundation.vertx.ClientEvent;
-import org.apache.servicecomb.foundation.vertx.ConnectionEvent;
-import org.apache.servicecomb.foundation.vertx.TransportType;
import org.apache.servicecomb.foundation.vertx.VertxTLSBuilder;
+import
org.apache.servicecomb.foundation.vertx.metrics.DefaultHttpServerMetrics;
+import
org.apache.servicecomb.foundation.vertx.metrics.metric.DefaultServerEndpointMetric;
import org.apache.servicecomb.swagger.invocation.exception.CommonExceptionData;
import org.apache.servicecomb.swagger.invocation.exception.InvocationException;
import
org.apache.servicecomb.transport.rest.vertx.accesslog.AccessLogConfiguration;
@@ -59,6 +54,7 @@
import io.vertx.core.http.HttpServer;
import io.vertx.core.http.HttpServerOptions;
import io.vertx.core.http.HttpServerResponse;
+import io.vertx.core.net.impl.ConnectionBase;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.RoutingContext;
import io.vertx.ext.web.handler.CorsHandler;
@@ -72,16 +68,6 @@
private URIEndpointObject endpointObject;
- private final AtomicInteger connectedCounter;
-
- public RestServerVerticle() {
-
this(CseContext.getInstance().getTransportManager().findTransport(Const.RESTFUL).getConnectedCounter());
- }
-
- public RestServerVerticle(AtomicInteger connectedCounter) {
- this.connectedCounter = connectedCounter;
- }
-
@Override
public void init(Vertx vertx, Context context) {
super.init(vertx, context);
@@ -107,17 +93,14 @@ public void start(Future<Void> startFuture) throws
Exception {
HttpServer httpServer = createHttpServer();
httpServer.requestHandler(mainRouter::accept);
httpServer.connectionHandler(connection -> {
- int connectedCount = connectedCounter.incrementAndGet();
+ DefaultHttpServerMetrics serverMetrics = (DefaultHttpServerMetrics)
((ConnectionBase) connection).metrics();
+ DefaultServerEndpointMetric endpointMetric =
serverMetrics.getEndpointMetric();
+ long connectedCount = endpointMetric.getCurrentConnectionCount();
int connectionLimit = DynamicPropertyFactory.getInstance()
.getIntProperty("servicecomb.rest.server.connection-limit",
Integer.MAX_VALUE).get();
if (connectedCount > connectionLimit) {
- connectedCounter.decrementAndGet();
connection.close();
- } else {
- EventManager.post(new
ClientEvent(connection.remoteAddress().toString(),
- ConnectionEvent.Connected, TransportType.Rest, connectedCount));
- connection.closeHandler(event -> EventManager.post(new
ClientEvent(connection.remoteAddress().toString(),
- ConnectionEvent.Closed, TransportType.Rest,
connectedCounter.decrementAndGet())));
+ endpointMetric.onRejectByConnectionLimit();
}
});
httpServer.exceptionHandler(e -> {
diff --git
a/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/TestRestServerVerticle.java
b/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/TestRestServerVerticle.java
index 0ef37e350..c9846cf35 100644
---
a/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/TestRestServerVerticle.java
+++
b/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/TestRestServerVerticle.java
@@ -62,7 +62,7 @@
@Before
public void setUp() {
- instance = new RestServerVerticle(new AtomicInteger());
+ instance = new RestServerVerticle();
startFuture = Future.future();
CseContext.getInstance().setTransportManager(new TransportManager());
@@ -95,7 +95,7 @@ public void testRestServerVerticleWithRouter(@Mocked
Transport transport, @Mocke
result = endpiont;
}
};
- RestServerVerticle server = new RestServerVerticle(new AtomicInteger());
+ RestServerVerticle server = new RestServerVerticle();
// process stuff done by Expectations
server.init(vertx, context);
server.start(startFuture);
@@ -122,7 +122,7 @@ public void testRestServerVerticleWithRouterSSL(@Mocked
Transport transport, @Mo
result = endpiont;
}
};
- RestServerVerticle server = new RestServerVerticle(new AtomicInteger());
+ RestServerVerticle server = new RestServerVerticle();
// process stuff done by Expectations
server.init(vertx, context);
server.start(startFuture);
@@ -149,7 +149,7 @@ public void testRestServerVerticleWithHttp2(@Mocked
Transport transport, @Mocked
result = endpiont;
}
};
- RestServerVerticle server = new RestServerVerticle(new AtomicInteger());
+ RestServerVerticle server = new RestServerVerticle();
boolean status = false;
try {
server.init(vertx, context);
@@ -245,7 +245,7 @@ CorsHandler getCorsHandler(String corsAllowedOrigin) {
Router router = Mockito.mock(Router.class);
Mockito.when(router.route()).thenReturn(Mockito.mock(Route.class));
- RestServerVerticle server = new RestServerVerticle(new AtomicInteger());
+ RestServerVerticle server = new RestServerVerticle();
Deencapsulation.invoke(server, "mountCorsHandler", router);
Assert.assertEquals(7, counter.get());
@@ -272,7 +272,7 @@ Route
handler(io.vertx.core.Handler<io.vertx.ext.web.RoutingContext> requestHand
Mockito.when(mainRouter.route()).thenReturn(route);
- RestServerVerticle restServerVerticle = new RestServerVerticle(new
AtomicInteger());
+ RestServerVerticle restServerVerticle = new RestServerVerticle();
Deencapsulation.invoke(restServerVerticle,
"mountGlobalRestFailureHandler", mainRouter);
Assert.assertNotNull(handlerHolder.value);
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> switch max connection limitation to metrics mechanism
> -----------------------------------------------------
>
> Key: SCB-1023
> URL: https://issues.apache.org/jira/browse/SCB-1023
> Project: Apache ServiceComb
> Issue Type: Task
> Components: Java-Chassis
> Reporter: wujimin
> Assignee: wujimin
> Priority: Major
> Fix For: java-chassis-1.1.0
>
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
