fizzgate/fizz-gateway-node
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: global concurrents -1

Browse Source
This commit is contained in:
hongqiaowei
2021-01-15 10:56:18 +08:00
parent 7745c63ce3
commit 5843dcc651
5 changed files with 106 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/we/filter/AbsFlowControlFilter.java
View File

@@ -42,7 +42,7 @@ import javax.annotation.Resource;
public abstract class AbsFlowControlFilter extends ProxyAggrFilter { public abstract class AbsFlowControlFilter extends ProxyAggrFilter {
private static final Logger log = LoggerFactory.getLogger(AbsFlowControlFilter.class); protected static final Logger log = LoggerFactory.getLogger(AbsFlowControlFilter.class);
protected static final String exceed = " exceed "; protected static final String exceed = " exceed ";
protected static final String concurrents = " concurrents "; protected static final String concurrents = " concurrents ";

65
src/main/java/we/filter/GlobalFlowControlFilter.java
View File

@@ -22,7 +22,14 @@ import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange; import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.WebFilterChain; import org.springframework.web.server.WebFilterChain;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;
import we.flume.clients.log4j2appender.LogService;
import we.stats.ratelimit.ResourceRateLimitConfig; import we.stats.ratelimit.ResourceRateLimitConfig;
import we.util.JacksonUtils;
import we.util.WebUtils;
import java.util.HashMap;
import java.util.Map;
/** /**
* @author hongqiaowei * @author hongqiaowei
@@ -38,34 +45,66 @@ public class GlobalFlowControlFilter extends AbsFlowControlFilter {
public Mono<Void> doFilter(ServerWebExchange exchange, WebFilterChain chain) { public Mono<Void> doFilter(ServerWebExchange exchange, WebFilterChain chain) {
if (flowControl) { if (flowControl) {
// Map<String, Object> traceMap = new HashMap<>();
LogService.setBizId(exchange.getRequest().getId());
long currentTimeSlot = flowStat.currentTimeSlotId(); long currentTimeSlot = flowStat.currentTimeSlotId();
// traceMap.put("currentTimeSlot", currentTimeSlot);
exchange.getAttributes().put(AbsFlowControlFilter.currentTimeSlot, currentTimeSlot); exchange.getAttributes().put(AbsFlowControlFilter.currentTimeSlot, currentTimeSlot);
ResourceRateLimitConfig config = resourceRateLimitConfigService.getResourceRateLimitConfig(ResourceRateLimitConfig.GLOBAL); ResourceRateLimitConfig config = resourceRateLimitConfigService.getResourceRateLimitConfig(ResourceRateLimitConfig.GLOBAL);
if (config.isEnable()) { if (config.isEnable()) {
// traceMap.put("globalConfig", "enable conns " + config.concurrents + " and incr now");
boolean concurrentOrRpsExceed = !flowStat.incrRequest(ResourceRateLimitConfig.GLOBAL, currentTimeSlot, config.concurrents, config.qps); boolean concurrentOrRpsExceed = !flowStat.incrRequest(ResourceRateLimitConfig.GLOBAL, currentTimeSlot, config.concurrents, config.qps);
if (concurrentOrRpsExceed) { if (concurrentOrRpsExceed) {
// traceMap.put("globalConfigExceed", "true");
return generateExceedResponse(exchange, config); return generateExceedResponse(exchange, config);
} }
} else { } else {
// traceMap.put("noGlobalConfig", "incr now");
flowStat.incrRequest(ResourceRateLimitConfig.GLOBAL, currentTimeSlot, null, null); flowStat.incrRequest(ResourceRateLimitConfig.GLOBAL, currentTimeSlot, null, null);
} }
// if (log.isDebugEnabled()) {
// log.debug(JacksonUtils.writeValueAsString(traceMap), LogService.BIZ_ID, exchange.getRequest().getId());
// }
// StringBuilder b = new StringBuilder();
// WebUtils.request2stringBuilder(exchange, b);
// b.append('\n');
long start = System.currentTimeMillis(); long start = System.currentTimeMillis();
exchange.getAttributes().put(AbsFlowControlFilter.start, start); exchange.getAttributes().put(AbsFlowControlFilter.start, start);
return chain.filter(exchange) return chain.filter(exchange)
.doOnSuccess( // .doOnSuccess(
r -> { // r -> {
inTheEnd(exchange, ResourceRateLimitConfig.GLOBAL, start, currentTimeSlot, true); // // b.append(" succ ");
} // // inTheEnd(exchange, ResourceRateLimitConfig.GLOBAL, start, currentTimeSlot, true);
) // }
.doOnError( // )
t -> { // .doOnError(
inTheEnd(exchange, ResourceRateLimitConfig.GLOBAL, start, currentTimeSlot, false); // t -> {
} // // b.append(" errs ");
) // // inTheEnd(exchange, ResourceRateLimitConfig.GLOBAL, start, currentTimeSlot, false);
.doOnCancel( // }
() -> { // )
inTheEnd(exchange, ResourceRateLimitConfig.GLOBAL, start, currentTimeSlot, false); // .doOnCancel(
// () -> {
// // b.append(" cans ");
// // inTheEnd(exchange, ResourceRateLimitConfig.GLOBAL, start, currentTimeSlot, false);
// }
// )
.doFinally(
s -> {
if (s == SignalType.ON_COMPLETE) {
// b.append(" comps ");
inTheEnd(exchange, ResourceRateLimitConfig.GLOBAL, start, currentTimeSlot, true);
} else {
// b.append(" " + s);
inTheEnd(exchange, ResourceRateLimitConfig.GLOBAL, start, currentTimeSlot, false);
}
// if (log.isDebugEnabled()) {
// log.debug(b.toString(), LogService.BIZ_ID, exchange.getRequest().getId());
// }
} }
); );
} }

36
src/main/java/we/filter/ServiceFlowControlFilter.java
View File

@@ -22,6 +22,7 @@ import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange; import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.WebFilterChain; import org.springframework.web.server.WebFilterChain;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;
import we.stats.ratelimit.ResourceRateLimitConfig; import we.stats.ratelimit.ResourceRateLimitConfig;
import we.util.WebUtils; import we.util.WebUtils;
@@ -57,19 +58,28 @@ public class ServiceFlowControlFilter extends AbsFlowControlFilter {
long start = exchange.getAttribute(AbsFlowControlFilter.start); long start = exchange.getAttribute(AbsFlowControlFilter.start);
return chain.filter(exchange) return chain.filter(exchange)
.doOnSuccess( // .doOnSuccess(
r -> { // r -> {
inTheEnd(exchange, service, start, currentTimeSlot, true); // inTheEnd(exchange, service, start, currentTimeSlot, true);
} // }
) // )
.doOnError( // .doOnError(
t -> { // t -> {
inTheEnd(exchange, service, start, currentTimeSlot, false); // inTheEnd(exchange, service, start, currentTimeSlot, false);
} // }
) // )
.doOnCancel( // .doOnCancel(
() -> { // () -> {
inTheEnd(exchange, service, start, currentTimeSlot, false); // inTheEnd(exchange, service, start, currentTimeSlot, false);
// }
// )
.doFinally(
s -> {
if (s == SignalType.ON_COMPLETE) {
inTheEnd(exchange, service, start, currentTimeSlot, true);
} else {
inTheEnd(exchange, service, start, currentTimeSlot, false);
}
} }
); );
} }

15
src/main/java/we/stats/FlowStat.java
View File

@@ -29,6 +29,7 @@ import java.util.concurrent.Executors;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import we.util.Utils;
/** /**
* Flow Statistic * Flow Statistic
@@ -132,6 +133,17 @@ public class FlowStat {
return; return;
} }
ResourceStat resourceStat = getResourceStat(resourceId); ResourceStat resourceStat = getResourceStat(resourceId);
long conns = resourceStat.getConcurrentRequests().get();
if (conns == 0) {
if (log.isDebugEnabled()) {
StringBuilder b = new StringBuilder(256);
b.append(timeSlotId + " " + resourceId + " conns 0 before decr it").append('\n');
Utils.threadCurrentStack2stringBuilder(b);
log.debug(b.toString());
}
}
resourceStat.decrConcurrentRequest(timeSlotId); resourceStat.decrConcurrentRequest(timeSlotId);
} }
@@ -158,6 +170,9 @@ public class FlowStat {
resourceStat = resourceStats.get(resourceId); resourceStat = resourceStats.get(resourceId);
} else { } else {
resourceStat = new ResourceStat(resourceId); resourceStat = new ResourceStat(resourceId);
if (log.isDebugEnabled()) {
log.debug("no resource stat for " + resourceId + ", create one " + resourceStat);
}
ResourceStat rs = resourceStats.putIfAbsent(resourceId, resourceStat); ResourceStat rs = resourceStats.putIfAbsent(resourceId, resourceStat);
if (rs != null) { if (rs != null) {
resourceStat = rs; resourceStat = rs;

24
src/test/java/we/stats/ratelimit/RateLimitTests.java
View File

@@ -52,7 +52,7 @@ public class RateLimitTests {
private ConnectionProvider getConnectionProvider() { private ConnectionProvider getConnectionProvider() {
return ConnectionProvider return ConnectionProvider
.builder("flow-control-cp") .builder("flow-control-cp")
.maxConnections(100) .maxConnections(500)
.pendingAcquireTimeout(Duration.ofMillis(6_000)) .pendingAcquireTimeout(Duration.ofMillis(6_000))
.maxIdleTime(Duration.ofMillis(40_000)) .maxIdleTime(Duration.ofMillis(40_000))
.build(); .build();
@@ -96,12 +96,15 @@ public class RateLimitTests {
} }
@Test @Test
public void flowControlTests() { public void flowControlTests() throws InterruptedException {
WebClient webClient = getWebClient(); WebClient webClient = getWebClient();
for (int i = 0; i < 10; i++) { for (int i = 0; i < 100_000; i++) {
// String uri = "http://12.5.3.8:8600/proxy/fizz" + i + "/fntrol/mock";
String uri = "http://12.5.3.8:8600/proxy/fizz/fntrol/mock" + i;
System.err.println(i);
webClient webClient
.method(HttpMethod.GET) .method(HttpMethod.GET)
.uri("") .uri(uri)
.headers(hdrs -> {}) .headers(hdrs -> {})
.body(Mono.just(""), String.class) .body(Mono.just(""), String.class)
.exchange().name("") .exchange().name("")
@@ -110,19 +113,22 @@ public class RateLimitTests {
.doOnError(t -> { .doOnError(t -> {
t.printStackTrace(); t.printStackTrace();
}) })
.timeout(Duration.ofMillis(6_000)) .timeout(Duration.ofMillis(16_000))
.flatMap( .flatMap(
remoteResp -> { remoteResp -> {
remoteResp.bodyToMono(String.class) remoteResp.bodyToMono(String.class)
.doOnSuccess( .doOnSuccess(
s -> { s -> {
System.out.println(s); // System.out.println(s);
} }
); );
return Mono.empty(); return Mono.empty();
} }
); )
.subscribe()
;
} }
Thread.currentThread().join();
} }
@Test @Test
@@ -130,12 +136,12 @@ public class RateLimitTests {
FlowStat flowStat = new FlowStat(); FlowStat flowStat = new FlowStat();
long incrTime = DateTimeUtils.toMillis("2021-01-08 21:28:42.000", Constants.DatetimePattern.DP23); long incrTime = DateTimeUtils.toMillis("2021-01-08 21:28:42.000", Constants.DatetimePattern.DP23);
boolean success = flowStat.incrRequest("resourceX", incrTime, Long.MAX_VALUE, Long.MAX_VALUE); boolean success = flowStat.incrRequest("resourceX", incrTime, Long.MAX_VALUE, Long.MAX_VALUE);
System.err.println("incrTime: " + incrTime + ", success: " + success); System.err.println("incrTime: " + incrTime + ", success: " + success);
long startTimeSlot = DateTimeUtils.toMillis("2021-01-08 21:28:41.000", Constants.DatetimePattern.DP23); long startTimeSlot = DateTimeUtils.toMillis("2021-01-08 21:28:41.000", Constants.DatetimePattern.DP23);
long endTimeSlot = DateTimeUtils.toMillis("2021-01-08 21:28:44.000", Constants.DatetimePattern.DP23); long endTimeSlot = DateTimeUtils.toMillis("2021-01-08 21:28:44.000", Constants.DatetimePattern.DP23);
List<ResourceTimeWindowStat> resourceTimeWindowStats = flowStat.getResourceTimeWindowStats(null, startTimeSlot, endTimeSlot, 3); List<ResourceTimeWindowStat> resourceTimeWindowStats = flowStat.getResourceTimeWindowStats(null, startTimeSlot, endTimeSlot, 3);
if (resourceTimeWindowStats == null || resourceTimeWindowStats.isEmpty()) { if (resourceTimeWindowStats == null || resourceTimeWindowStats.isEmpty()) {