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

Merge pull request #448 from wehotel/gray-release

Browse Source
This commit is contained in:
hongqiaowei
2022-07-28 11:34:11 +08:00
committed by GitHub
parent f2c893bcf3 23d77287b7
commit 64fbf0191b
22 changed files with 1100 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
fizz-bootstrap/pom.xml
View File

@@ -6,7 +6,7 @@
<parent> <parent>
<artifactId>fizz-gateway-community</artifactId> <artifactId>fizz-gateway-community</artifactId>
<groupId>com.fizzgate</groupId> <groupId>com.fizzgate</groupId>
<version>2.6.6</version> <version>2.6.7-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
@@ -32,7 +32,7 @@
<commons-codec.version>1.15</commons-codec.version> <commons-codec.version>1.15</commons-codec.version>
<commons-pool2.version>2.11.1</commons-pool2.version> <commons-pool2.version>2.11.1</commons-pool2.version>
<gson.version>2.8.9</gson.version> <gson.version>2.8.9</gson.version>
<netty-tcnative.version>2.0.53.Final</netty-tcnative.version> <netty-tcnative.version>2.0.54.Final</netty-tcnative.version>
<spring-cloud.version>2.2.9.RELEASE</spring-cloud.version> <spring-cloud.version>2.2.9.RELEASE</spring-cloud.version>
<snakeyaml.version>1.30</snakeyaml.version> <snakeyaml.version>1.30</snakeyaml.version>
<spring-data-releasetrain.version>Moore-SR13</spring-data-releasetrain.version>--> <spring-data-releasetrain.version>Moore-SR13</spring-data-releasetrain.version>-->

12
fizz-common/pom.xml
View File

@@ -5,7 +5,7 @@
<parent> <parent>
<artifactId>fizz-gateway-community</artifactId> <artifactId>fizz-gateway-community</artifactId>
<groupId>com.fizzgate</groupId> <groupId>com.fizzgate</groupId>
<version>2.6.6</version> <version>2.6.7-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>
@@ -17,6 +17,16 @@
</properties> </properties>
<dependencies> <dependencies>
<dependency>
<groupId>com.github.seancfoley</groupId>
<artifactId>ipaddress</artifactId>
</dependency>
<dependency>
<groupId>ognl</groupId>
<artifactId>ognl</artifactId>
</dependency>
<dependency> <dependency>
<groupId>org.openjdk.jol</groupId> <groupId>org.openjdk.jol</groupId>
<artifactId>jol-core</artifactId> <artifactId>jol-core</artifactId>

6
fizz-common/src/main/java/we/spring/http/server/reactive/ext/FizzServerHttpRequestDecorator.java
View File

@@ -68,7 +68,10 @@ public class FizzServerHttpRequestDecorator extends ServerHttpRequestDecorator {
public FizzServerHttpRequestDecorator(ServerHttpRequest delegate) { public FizzServerHttpRequestDecorator(ServerHttpRequest delegate) {
super(delegate); super(delegate);
this.delegate = (AbstractServerHttpRequest) delegate; this.delegate = (AbstractServerHttpRequest) delegate;
try {
nativeRequest = this.delegate.getNativeRequest(); nativeRequest = this.delegate.getNativeRequest();
} catch (IllegalStateException e) {
}
} }
@Override @Override
@@ -120,6 +123,9 @@ public class FizzServerHttpRequestDecorator extends ServerHttpRequestDecorator {
} }
private MultiValueMap<String, HttpCookie> initCookies() { private MultiValueMap<String, HttpCookie> initCookies() {
if (nativeRequest == null) {
return null;
}
MultiValueMap<String, HttpCookie> cookies = new LinkedMultiValueMap<>(); MultiValueMap<String, HttpCookie> cookies = new LinkedMultiValueMap<>();
for (CharSequence name : nativeRequest.cookies().keySet()) { for (CharSequence name : nativeRequest.cookies().keySet()) {
for (Cookie cookie : nativeRequest.cookies().get(name)) { for (Cookie cookie : nativeRequest.cookies().get(name)) {

47
fizz-common/src/test/java/we/OgnlTests.java Normal file
View File

@@ -0,0 +1,47 @@
package we;
import ognl.Ognl;
import org.junit.jupiter.api.Test;
import java.util.HashMap;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
class OgnlTests {
@Test
void testGet() throws Exception {
Root root = new Root();
Map<String, Object> query = new HashMap<>();
query.put("version", "v2");
query.put("userId", 1234563);
query.put("age", 25);
root.put("query", query);
Map<String, Object> client = new HashMap<>();
client.put("ip", "10.2.3.4");
client.put("ip2", "10.2.3.88");
root.put("client", client);
Boolean result = (Boolean) Ognl.getValue("checkIp(client.ip2) && (query.version == 'v2' || query.age < 20) and query.age in (22,25,30) && client.ip=='10.2.3.4'", root);
System.out.println(result);
assertEquals(true, result);
}
}
class Root extends HashMap {
public Root() {
}
public boolean checkIp(String ip) {
System.out.println(ip);
return ip.equals("10.2.3.88");
}
}

2
fizz-core/pom.xml
View File

@@ -5,7 +5,7 @@
<parent> <parent>
<artifactId>fizz-gateway-community</artifactId> <artifactId>fizz-gateway-community</artifactId>
<groupId>com.fizzgate</groupId> <groupId>com.fizzgate</groupId>
<version>2.6.6</version> <version>2.6.7-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

1
fizz-core/src/main/java/we/filter/AggregateFilter.java
View File

@@ -83,6 +83,7 @@ public class AggregateFilter implements WebFilter {
String serviceId = WebUtils.getBackendService(exchange); String serviceId = WebUtils.getBackendService(exchange);
if (serviceId == null) { if (serviceId == null) {
return chain.filter(exchange); return chain.filter(exchange);
} else if (WebUtils.ignorePlugin(exchange) && WebUtils.getRoute(exchange).type == ApiConfig.Type.SERVICE_AGGREGATE) {
} else { } else {
byte act = WebUtils.getApiConfigType(exchange); byte act = WebUtils.getApiConfigType(exchange);
if (act == ApiConfig.Type.UNDEFINED) { if (act == ApiConfig.Type.UNDEFINED) {

12
fizz-core/src/main/java/we/filter/FlowControlFilter.java
View File

@@ -168,7 +168,7 @@ public class FlowControlFilter extends FizzWebFilter {
long currentTimeMillis = System.currentTimeMillis(); long currentTimeMillis = System.currentTimeMillis();
String blockedResourceId = result.getBlockedResourceId(); String blockedResourceId = result.getBlockedResourceId();
if (BlockType.CIRCUIT_BREAK == result.getBlockType()) { if (BlockType.CIRCUIT_BREAK == result.getBlockType()) {
fizzMonitorService.sendAlarm(service, path, FizzMonitorService.CIRCUIT_BREAK_ALARM, null, currentTimeMillis); fizzMonitorService.alarm(service, path, FizzMonitorService.CIRCUIT_BREAK_ALARM, null);
// log.info("{} trigger {} circuit breaker limit", traceId, blockedResourceId, LogService.BIZ_ID, traceId); // log.info("{} trigger {} circuit breaker limit", traceId, blockedResourceId, LogService.BIZ_ID, traceId);
log.info("{} trigger {} circuit breaker limit", traceId, blockedResourceId); log.info("{} trigger {} circuit breaker limit", traceId, blockedResourceId);
@@ -200,11 +200,11 @@ public class FlowControlFilter extends FizzWebFilter {
} else { } else {
if (BlockType.CONCURRENT_REQUEST == result.getBlockType()) { if (BlockType.CONCURRENT_REQUEST == result.getBlockType()) {
fizzMonitorService.sendAlarm(service, path, FizzMonitorService.RATE_LIMIT_ALARM, concurrents, currentTimeMillis); fizzMonitorService.alarm(service, path, FizzMonitorService.RATE_LIMIT_ALARM, concurrents);
// log.info("{} exceed {} flow limit, blocked by maximum concurrent requests", traceId, blockedResourceId, LogService.BIZ_ID, traceId); // log.info("{} exceed {} flow limit, blocked by maximum concurrent requests", traceId, blockedResourceId, LogService.BIZ_ID, traceId);
log.info("{} exceed {} flow limit, blocked by maximum concurrent requests", traceId, blockedResourceId); log.info("{} exceed {} flow limit, blocked by maximum concurrent requests", traceId, blockedResourceId);
} else { } else {
fizzMonitorService.sendAlarm(service, path, FizzMonitorService.RATE_LIMIT_ALARM, qps, currentTimeMillis); fizzMonitorService.alarm(service, path, FizzMonitorService.RATE_LIMIT_ALARM, qps);
// log.info("{} exceed {} flow limit, blocked by maximum QPS", traceId, blockedResourceId, LogService.BIZ_ID, traceId); // log.info("{} exceed {} flow limit, blocked by maximum QPS", traceId, blockedResourceId, LogService.BIZ_ID, traceId);
log.info("{} exceed {} flow limit, blocked by maximum QPS", traceId, blockedResourceId); log.info("{} exceed {} flow limit, blocked by maximum QPS", traceId, blockedResourceId);
} }
@@ -246,11 +246,11 @@ public class FlowControlFilter extends FizzWebFilter {
cb.transit(CircuitBreaker.State.RESUME_DETECTIVE, CircuitBreaker.State.OPEN, currentTimeSlot, flowStat); cb.transit(CircuitBreaker.State.RESUME_DETECTIVE, CircuitBreaker.State.OPEN, currentTimeSlot, flowStat);
} }
if (statusCode == HttpStatus.GATEWAY_TIMEOUT) { if (statusCode == HttpStatus.GATEWAY_TIMEOUT) {
fizzMonitorService.sendAlarm(finalService, finalPath, FizzMonitorService.TIMEOUT_ALARM, t.getMessage(), start); fizzMonitorService.alarm(finalService, finalPath, FizzMonitorService.TIMEOUT_ALARM, t.getMessage());
} else if (statusCode.is5xxServerError()) { } else if (statusCode.is5xxServerError()) {
fizzMonitorService.sendAlarm(finalService, finalPath, FizzMonitorService.ERROR_ALARM, String.valueOf(statusCode.value()), start); fizzMonitorService.alarm(finalService, finalPath, FizzMonitorService.ERROR_ALARM, String.valueOf(statusCode.value()));
} else if (s == SignalType.ON_ERROR && t != null) { } else if (s == SignalType.ON_ERROR && t != null) {
fizzMonitorService.sendAlarm(finalService, finalPath, FizzMonitorService.ERROR_ALARM, t.getMessage(), start); fizzMonitorService.alarm(finalService, finalPath, FizzMonitorService.ERROR_ALARM, t.getMessage());
} }
} else { } else {
flowStat.addRequestRT(resourceConfigs, currentTimeSlot, rt, true, statusCode); flowStat.addRequestRT(resourceConfigs, currentTimeSlot, rt, true, statusCode);

134
fizz-core/src/main/java/we/monitor/FizzMonitorService.java
View File

@@ -21,32 +21,50 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate; import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import we.config.AggregateRedisConfig; import we.config.AggregateRedisConfig;
import we.config.SchedConfig;
import we.util.Consts; import we.util.Consts;
import we.util.DateTimeUtils;
import we.util.JacksonUtils;
import we.util.ThreadContext; import we.util.ThreadContext;
import javax.annotation.Resource; import javax.annotation.Resource;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
/** /**
* @author hongqiaowei * @author hongqiaowei
*/ */
@Service @Service
public class FizzMonitorService { public class FizzMonitorService extends SchedConfig {
private static final Logger LOGGER = LoggerFactory.getLogger("monitor"); private static final Logger MONITOR_LOGGER = LoggerFactory.getLogger("monitor");
private static final Logger LOGGER = LoggerFactory.getLogger(FizzMonitorService.class);
public static final byte ERROR_ALARM = 1; public static final byte ERROR_ALARM = 1;
public static final byte TIMEOUT_ALARM = 2; public static final byte TIMEOUT_ALARM = 2;
public static final byte RATE_LIMIT_ALARM = 3; public static final byte RATE_LIMIT_ALARM = 3;
public static final byte CIRCUIT_BREAK_ALARM = 4; public static final byte CIRCUIT_BREAK_ALARM = 4;
private static final String _service = "\"service\":"; private static class Alarm {
private static final String _path = "\"path\":";
private static final String _type = "\"type\":"; public String service;
private static final String _desc = "\"desc\":"; public String path;
private static final String _timestamp = "\"timestamp\":"; public int type;
public String desc;
public long timestamp;
public int reqs = 0;
public long start;
@Override
public String toString() {
return JacksonUtils.writeValueAsString(this);
}
}
@Value("${fizz.monitor.alarm.enable:true}") @Value("${fizz.monitor.alarm.enable:true}")
private boolean alarmEnable; private boolean alarmEnable;
@@ -54,38 +72,102 @@ public class FizzMonitorService {
@Value("${fizz.monitor.alarm.dest:redis}") @Value("${fizz.monitor.alarm.dest:redis}")
private String dest; private String dest;
@Value("${fizz.monitor.alarm.queue:fizz_alarm_channel}") @Value("${fizz.monitor.alarm.queue:fizz_alarm_channel_new}")
private String queue; private String queue;
@Resource(name = AggregateRedisConfig.AGGREGATE_REACTIVE_REDIS_TEMPLATE) @Resource(name = AggregateRedisConfig.AGGREGATE_REACTIVE_REDIS_TEMPLATE)
private ReactiveStringRedisTemplate rt; private ReactiveStringRedisTemplate rt;
public void sendAlarm(String service, String path, byte type, String desc, long timestamp) { private Map<Long/*thread id*/,
if (alarmEnable) { Map/*LinkedHashMap*/<Long/*time win start*/,
StringBuilder b = ThreadContext.getStringBuilder(); Map<String/*service+path+type*/, Alarm>
b.append(Consts.S.LEFT_BRACE); >
b.append(_service); toJsonStrVal(b, service); b.append(Consts.S.COMMA); >
b.append(_path); toJsonStrVal(b, path); b.append(Consts.S.COMMA); threadTimeWinAlarmMap = new HashMap<>();
b.append(_type); b.append(type); b.append(Consts.S.COMMA);
if (desc != null) { public void alarm(String service, String path, byte type, String desc) {
b.append(_desc); toJsonStrVal(b, desc); b.append(Consts.S.COMMA); if (alarmEnable) {
long tid = Thread.currentThread().getId();
Map<Long, Map<String, Alarm>> timeWinAlarmMap = threadTimeWinAlarmMap.get(tid);
if (timeWinAlarmMap == null) {
timeWinAlarmMap = new LinkedHashMap<Long, Map<String, Alarm>>(4, 1) {
@Override
protected boolean removeEldestEntry(Map.Entry eldest) {
return size() > 2;
}
};
threadTimeWinAlarmMap.put(tid, timeWinAlarmMap);
} }
b.append(_timestamp) .append(timestamp); long currentTimeWinStart = DateTimeUtils.get10sTimeWinStart(1);
b.append(Consts.S.RIGHT_BRACE); Map<String, Alarm> alarmMap = timeWinAlarmMap.computeIfAbsent(currentTimeWinStart, k -> new HashMap<>(128));
String msg = b.toString();
String key = ThreadContext.getStringBuilder().append(service).append(path).append(type).toString();
Alarm alarm = alarmMap.get(key);
if (alarm == null) {
alarm = new Alarm();
alarm.service = service;
alarm.path = path;
alarm.type = type;
alarmMap.put(key, alarm);
}
alarm.desc = desc;
alarm.timestamp = System.currentTimeMillis();
alarm.reqs++;
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("update alarm: {} at {}", alarm, DateTimeUtils.convert(alarm.timestamp, Consts.DP.DP19));
}
}
}
@Scheduled(cron = "${fizz.monitor.alarm.sched.cron:2/10 * * * * ?}")
public void sched() {
long prevTimeWinStart = DateTimeUtils.get10sTimeWinStart(2);
Map<String, Alarm> alarmMap = ThreadContext.getHashMap();
threadTimeWinAlarmMap.forEach(
(t, timeWinAlarmMap) -> {
Map<String, Alarm> alarmMap0 = timeWinAlarmMap.get(prevTimeWinStart);
if (alarmMap0 != null) {
alarmMap0.forEach(
(spt, alarm) -> {
Alarm a = alarmMap.get(spt);
if (a == null) {
alarm.start = prevTimeWinStart;
alarmMap.put(spt, alarm);
} else {
a.reqs = a.reqs + alarm.reqs;
if (alarm.timestamp > a.timestamp) {
a.timestamp = alarm.timestamp;
a.desc = alarm.desc;
}
}
}
);
}
}
);
if (alarmMap.isEmpty()) {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("no alarm in {} window", DateTimeUtils.convert(prevTimeWinStart, Consts.DP.DP19));
}
} else {
alarmMap.forEach(
(spt, alarm) -> {
String msg = alarm.toString();
if (Consts.KAFKA.equals(dest)) { if (Consts.KAFKA.equals(dest)) {
// LOGGER.warn(msg, LogService.HANDLE_STGY, LogService.toKF(queue)); MONITOR_LOGGER.info(msg);
LOGGER.info(msg); if (LOGGER.isDebugEnabled()) {
LOGGER.debug("send alarm {} which belong to {} window to topic", msg, DateTimeUtils.convert(alarm.start, Consts.DP.DP19));
}
} else { } else {
rt.convertAndSend(queue, msg).subscribe(); rt.convertAndSend(queue, msg).subscribe();
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("send alarm {} which belong to {} window to channel {}", msg, DateTimeUtils.convert(alarm.start, Consts.DP.DP19), queue);
} }
} }
} }
);
}
}
private static void toJsonStrVal(StringBuilder b, String value) {
b.append(Consts.S.DOUBLE_QUOTE).append(value).append(Consts.S.DOUBLE_QUOTE);
}
} }

4
fizz-core/src/main/java/we/plugin/FizzPluginFilterChain.java
View File

@@ -44,6 +44,10 @@ public final class FizzPluginFilterChain {
} }
public static Mono<Void> next(ServerWebExchange exchange) { public static Mono<Void> next(ServerWebExchange exchange) {
if (WebUtils.ignorePlugin(exchange)) {
WebFilterChain chain = exchange.getAttribute(WEB_FILTER_CHAIN);
return chain.filter(exchange);
}
Iterator<PluginConfig> it = exchange.getAttribute(pluginConfigsIt); Iterator<PluginConfig> it = exchange.getAttribute(pluginConfigsIt);
Route route = WebUtils.getRoute(exchange); Route route = WebUtils.getRoute(exchange);
if (it == null || route.pluginConfigsChange) { if (it == null || route.pluginConfigsChange) {

2
fizz-core/src/main/java/we/plugin/auth/ApiConfig.java
View File

@@ -47,6 +47,7 @@ public class ApiConfig {
static final byte REVERSE_PROXY = 3; static final byte REVERSE_PROXY = 3;
static final byte CALLBACK = 4; static final byte CALLBACK = 4;
static final byte DUBBO = 5; static final byte DUBBO = 5;
static final byte DIRECT_RESPONSE = 6;
} }
public static final String ALL_METHOD = "AM"; public static final String ALL_METHOD = "AM";
@@ -226,6 +227,7 @@ public class ApiConfig {
Route r = new Route().dedicatedLine( this.dedicatedLine) Route r = new Route().dedicatedLine( this.dedicatedLine)
.type( this.type) .type( this.type)
.method( request.getMethod()) .method( request.getMethod())
.path( this.path)
.registryCenter( this.registryCenter) .registryCenter( this.registryCenter)
.backendService( this.backendService) .backendService( this.backendService)
.backendPath( this.backendPath) .backendPath( this.backendPath)

168
fizz-core/src/main/java/we/plugin/stat/AccessStatSchedConfig.java
View File

@@ -1,84 +1,84 @@
///* /*
// * Copyright (C) 2021 the original author or authors. * Copyright (C) 2021 the original author or authors.
// * *
// * This program is free software: you can redistribute it and/or modify * This program is free software: you can redistribute it and/or modify
// * it under the terms of the GNU General Public License as published by * it under the terms of the GNU General Public License as published by
// * the Free Software Foundation, either version 3 of the License, or * the Free Software Foundation, either version 3 of the License, or
// * any later version. * any later version.
// * *
// * This program is distributed in the hope that it will be useful, * This program is distributed in the hope that it will be useful,
// * but WITHOUT ANY WARRANTY; without even the implied warranty of * but WITHOUT ANY WARRANTY; without even the implied warranty of
// * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// * GNU General Public License for more details. * GNU General Public License for more details.
// * *
// * You should have received a copy of the GNU General Public License * You should have received a copy of the GNU General Public License
// * along with this program. If not, see <https://www.gnu.org/licenses/>. * along with this program. If not, see <https://www.gnu.org/licenses/>.
// */ */
//
//package we.plugin.stat; package we.plugin.stat;
//
//import org.slf4j.Logger; import org.slf4j.Logger;
//import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
//import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Configuration;
//import org.springframework.data.redis.core.ReactiveStringRedisTemplate; import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
//import org.springframework.scheduling.annotation.Scheduled; import org.springframework.scheduling.annotation.Scheduled;
//import we.config.AggregateRedisConfig; import we.config.AggregateRedisConfig;
//import we.config.SchedConfig; import we.config.SchedConfig;
//import we.util.Consts; import we.util.Consts;
//import we.util.DateTimeUtils; import we.util.DateTimeUtils;
//import we.util.StringUtils; import we.util.StringUtils;
//
//import javax.annotation.Resource; import javax.annotation.Resource;
//import java.util.Map; import java.util.Map;
//
///** /**
// * @author hongqiaowei * @author hongqiaowei
// */ */
//
//@Configuration @Configuration
//public class AccessStatSchedConfig extends SchedConfig { public class AccessStatSchedConfig extends SchedConfig {
//
// private static final Logger LOGGER = LoggerFactory.getLogger(AccessStatSchedConfig.class); private static final Logger LOGGER = LoggerFactory.getLogger(AccessStatSchedConfig.class);
//
// private static final Logger STAT_LOGGER = LoggerFactory.getLogger("stat"); private static final Logger STAT_LOGGER = LoggerFactory.getLogger("stat");
//
// @Resource(name = AggregateRedisConfig.AGGREGATE_REACTIVE_REDIS_TEMPLATE) @Resource(name = AggregateRedisConfig.AGGREGATE_REACTIVE_REDIS_TEMPLATE)
// private ReactiveStringRedisTemplate rt; private ReactiveStringRedisTemplate rt;
//
// @Resource @Resource
// private StatPluginFilterProperties statPluginFilterProperties; private StatPluginFilterProperties statPluginFilterProperties;
//
// @Resource @Resource
// private StatPluginFilter statPluginFilter; private StatPluginFilter statPluginFilter;
//
// @Scheduled(cron = "${fizz-access-stat-sched.cron:2/10 * * * * ?}") @Scheduled(cron = "${fizz-access-stat-sched.cron:2/10 * * * * ?}")
// public void sched() { public void sched() {
// long prevTimeWinStart = DateTimeUtils.get10sTimeWinStart(2); long prevTimeWinStart = DateTimeUtils.get10sTimeWinStart(2);
// Map<String, AccessStat> accessStatMap = statPluginFilter.getAccessStat(prevTimeWinStart); Map<String, AccessStat> accessStatMap = statPluginFilter.getAccessStat(prevTimeWinStart);
//
// if (accessStatMap.isEmpty()) { if (accessStatMap.isEmpty()) {
// if (LOGGER.isDebugEnabled()) { if (LOGGER.isDebugEnabled()) {
// LOGGER.debug("no access stat in {} window", DateTimeUtils.convert(prevTimeWinStart, Consts.DP.DP19)); LOGGER.debug("no access stat in {} window", DateTimeUtils.convert(prevTimeWinStart, Consts.DP.DP19));
// } }
// } else { } else {
// accessStatMap.forEach( accessStatMap.forEach(
// (smp, accessStat) -> { (smp, accessStat) -> {
// String msg = accessStat.toString(); String msg = accessStat.toString();
// String topic = statPluginFilterProperties.getFizzAccessStatTopic(); String topic = statPluginFilterProperties.getFizzAccessStatTopic();
// if (StringUtils.isBlank(topic)) { if (StringUtils.isBlank(topic)) {
// String channel = statPluginFilterProperties.getFizzAccessStatChannel(); String channel = statPluginFilterProperties.getFizzAccessStatChannel();
// rt.convertAndSend(channel, msg).subscribe(); rt.convertAndSend(channel, msg).subscribe();
// if (LOGGER.isDebugEnabled()) { if (LOGGER.isDebugEnabled()) {
// LOGGER.debug("send access stat {} which belong to {} window to channel {}", msg, DateTimeUtils.convert(accessStat.start, Consts.DP.DP19), channel); LOGGER.debug("send access stat {} which belong to {} window to channel {}", msg, DateTimeUtils.convert(accessStat.start, Consts.DP.DP19), channel);
// } }
// } else { } else {
// STAT_LOGGER.info(msg); STAT_LOGGER.info(msg);
// if (LOGGER.isDebugEnabled()) { if (LOGGER.isDebugEnabled()) {
// LOGGER.debug("send access stat {} which belong to {} window to topic", msg, DateTimeUtils.convert(accessStat.start, Consts.DP.DP19)); LOGGER.debug("send access stat {} which belong to {} window to topic", msg, DateTimeUtils.convert(accessStat.start, Consts.DP.DP19));
// } }
// } }
// } }
// ); );
// } }
// } }
//} }

118
fizz-core/src/main/java/we/plugin/stat/StatPluginFilter.java
View File

@@ -17,107 +17,107 @@
package we.plugin.stat; package we.plugin.stat;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange; import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import we.config.AggregateRedisConfig;
import we.plugin.PluginFilter; import we.plugin.PluginFilter;
import we.plugin.auth.GatewayGroupService;
import we.util.Consts; import we.util.Consts;
import we.util.DateTimeUtils;
import we.util.ThreadContext; import we.util.ThreadContext;
import we.util.WebUtils; import we.util.WebUtils;
import javax.annotation.Resource; import javax.annotation.Resource;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map; import java.util.Map;
/** /**
* @author hongqiaowei * @author hongqiaowei
* @apiNote unstable.
*/ */
@Component(StatPluginFilter.STAT_PLUGIN_FILTER) @Component(StatPluginFilter.STAT_PLUGIN_FILTER)
public class StatPluginFilter extends PluginFilter { public class StatPluginFilter extends PluginFilter {
private static final Logger log = LoggerFactory.getLogger("stat"); private static final Logger LOGGER = LoggerFactory.getLogger(StatPluginFilter.class);
public static final String STAT_PLUGIN_FILTER = "statPlugin"; public static final String STAT_PLUGIN_FILTER = "statPlugin";
private static final String ip = "\"ip\":";
private static final String gatewayGroup = "\"gatewayGroup\":";
private static final String service = "\"service\":";
private static final String appid = "\"appid\":";
private static final String apiMethod = "\"apiMethod\":";
private static final String apiPath = "\"apiPath\":";
private static final String reqTime = "\"reqTime\":";
@Resource @Resource
private StatPluginFilterProperties statPluginFilterProperties; private StatPluginFilterProperties statPluginFilterProperties;
@Resource(name = AggregateRedisConfig.AGGREGATE_REACTIVE_REDIS_TEMPLATE) private Map<Long/*thread id*/,
private ReactiveStringRedisTemplate rt; Map/*LinkedHashMap*/<Long/*time win start*/,
Map<String/*service+apiMethod+apiPath*/, AccessStat>
@Resource >
private GatewayGroupService gatewayGroupService; >
threadTimeWinAccessStatMap = new HashMap<>();
@Override @Override
public Mono<Void> doFilter(ServerWebExchange exchange, Map<String, Object> config, String fixedConfig) { public Mono<Void> doFilter(ServerWebExchange exchange, Map<String, Object> config, String fixedConfig) {
if (statPluginFilterProperties.isStatOpen()) { if (statPluginFilterProperties.isStatOpen()) {
StringBuilder b = ThreadContext.getStringBuilder(); long tid = Thread.currentThread().getId();
b.append(Consts.S.LEFT_BRACE); Map<Long, Map<String, AccessStat>> timeWinAccessStatMap = threadTimeWinAccessStatMap.get(tid);
b.append(ip); toJsonStringValue(b, WebUtils.getOriginIp(exchange)); b.append(Consts.S.COMMA); if (timeWinAccessStatMap == null) {
b.append(gatewayGroup); toJsonStringValue(b, currentGatewayGroups()); b.append(Consts.S.COMMA); timeWinAccessStatMap = new LinkedHashMap<Long, Map<String, AccessStat>>(4, 1) {
b.append(service); toJsonStringValue(b, WebUtils.getClientService(exchange)); b.append(Consts.S.COMMA); @Override
protected boolean removeEldestEntry(Map.Entry eldest) {
String appId = WebUtils.getAppId(exchange); return size() > 2;
if (appId != null) { }
b.append(appid); toJsonStringValue(b, appId); b.append(Consts.S.COMMA); };
threadTimeWinAccessStatMap.put(tid, timeWinAccessStatMap);
} }
b.append(apiMethod); toJsonStringValue(b, exchange.getRequest().getMethodValue()); b.append(Consts.S.COMMA); long currentTimeWinStart = DateTimeUtils.get10sTimeWinStart(1);
b.append(apiPath); toJsonStringValue(b, WebUtils.getClientReqPath(exchange)); b.append(Consts.S.COMMA); Map<String, AccessStat> accessStatMap = timeWinAccessStatMap.computeIfAbsent(currentTimeWinStart, k -> new HashMap<>(128));
b.append(reqTime) .append(System.currentTimeMillis());
b.append(Consts.S.RIGHT_BRACE);
if (StringUtils.isBlank(statPluginFilterProperties.getFizzAccessStatTopic())) { String service = WebUtils.getClientService(exchange);
rt.convertAndSend(statPluginFilterProperties.getFizzAccessStatChannel(), b.toString()).subscribe(); String method = exchange.getRequest().getMethodValue();
} else { String path = WebUtils.getClientReqPath(exchange);
// log.warn(b.toString(), LogService.HANDLE_STGY, LogService.toKF(statPluginFilterProperties.getFizzAccessStatTopic())); // for internal use String key = ThreadContext.getStringBuilder().append(service).append(method).append(path).toString();
log.info(b.toString()); AccessStat accessStat = accessStatMap.get(key);
if (accessStat == null) {
accessStat = new AccessStat();
accessStat.service = service;
accessStat.apiMethod = method;
accessStat.apiPath = path;
accessStatMap.put(key, accessStat);
}
accessStat.reqTime = System.currentTimeMillis();
accessStat.reqs++;
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("update access stat: {}, which request at {}", accessStat, DateTimeUtils.convert(accessStat.reqTime, Consts.DP.DP19));
} }
} }
return WebUtils.transmitSuccessFilterResultAndEmptyMono(exchange, STAT_PLUGIN_FILTER, null); return WebUtils.transmitSuccessFilterResultAndEmptyMono(exchange, STAT_PLUGIN_FILTER, null);
} }
private String currentGatewayGroups() { public Map<String, AccessStat> getAccessStat(long timeWinStart) {
int sz = gatewayGroupService.currentGatewayGroupSet.size(); Map<String, AccessStat> result = ThreadContext.getHashMap();
if (sz == 1) { threadTimeWinAccessStatMap.forEach(
return gatewayGroupService.currentGatewayGroupSet.iterator().next(); (t, timeWinAccessStatMap) -> {
} Map<String, AccessStat> accessStatMap = timeWinAccessStatMap.get(timeWinStart);
StringBuilder b = ThreadContext.getStringBuilder(ThreadContext.sb0); if (accessStatMap != null) {
byte i = 0; accessStatMap.forEach(
for (String g : gatewayGroupService.currentGatewayGroupSet) { (smp, accessStat) -> {
b.append(g); AccessStat as = result.get(smp);
i++; if (as == null) {
if (i < sz) { accessStat.start = timeWinStart;
b.append(Consts.S.COMMA); result.put(smp, accessStat);
} else {
as.reqs = as.reqs + accessStat.reqs;
if (accessStat.reqTime > as.reqTime) {
as.reqTime = accessStat.reqTime;
} }
} }
return b.toString();
} }
);
private static void toJsonStringValue(StringBuilder b, String value) { }
b.append(Consts.S.DOUBLE_QUOTE).append(value).append(Consts.S.DOUBLE_QUOTE); }
);
return result;
} }
} }

2
fizz-core/src/main/java/we/plugin/stat/StatPluginFilterProperties.java
View File

@@ -37,7 +37,7 @@ public class StatPluginFilterProperties {
@Value("${stat.open:false}") @Value("${stat.open:false}")
private boolean statOpen = false; private boolean statOpen = false;
@Value("${stat.channel:fizz_access_stat}") @Value("${stat.channel:fizz_access_stat_new}")
private String fizzAccessStatChannel; private String fizzAccessStatChannel;
@Value("${stat.topic:}") @Value("${stat.topic:}")

22
fizz-core/src/main/java/we/proxy/Route.java
View File

@@ -18,6 +18,7 @@
package we.proxy; package we.proxy;
import org.springframework.http.HttpMethod; import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import we.plugin.PluginConfig; import we.plugin.PluginConfig;
import we.util.Consts; import we.util.Consts;
import we.util.JacksonUtils; import we.util.JacksonUtils;
@@ -36,6 +37,8 @@ public class Route {
public HttpMethod method; public HttpMethod method;
public String path;
public String registryCenter; public String registryCenter;
public String backendService; public String backendService;
@@ -68,6 +71,10 @@ public class Route {
public long retryInterval = 0; public long retryInterval = 0;
public MediaType contentType;
public String body;
public Route dedicatedLine(boolean b) { public Route dedicatedLine(boolean b) {
dedicatedLine = b; dedicatedLine = b;
return this; return this;
@@ -83,6 +90,11 @@ public class Route {
return this; return this;
} }
public Route path(String p) {
path = p;
return this;
}
public Route registryCenter(String rc) { public Route registryCenter(String rc) {
registryCenter = rc; registryCenter = rc;
return this; return this;
@@ -150,6 +162,16 @@ public class Route {
return this; return this;
} }
public Route contentType(MediaType type) {
contentType = type;
return this;
}
public Route body(String b) {
body = b;
return this;
}
@Deprecated @Deprecated
public String getBackendPathQuery() { public String getBackendPathQuery() {
if (query != null) { if (query != null) {

11
fizz-core/src/main/java/we/service_registry/RegistryCenterService.java
View File

@@ -28,10 +28,7 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import we.config.AggregateRedisConfig; import we.config.AggregateRedisConfig;
import we.config.SystemConfig; import we.config.SystemConfig;
import we.util.Consts; import we.util.*;
import we.util.JacksonUtils;
import we.util.Result;
import we.util.ThreadContext;
import javax.annotation.Resource; import javax.annotation.Resource;
import java.util.Collections; import java.util.Collections;
@@ -182,7 +179,11 @@ public class RegistryCenterService implements ApplicationListener<ContextRefresh
} }
public String getInstance(String registryCenter, String service) { public String getInstance(String registryCenter, String service) {
return registryCenterMap.get(registryCenter).getInstance(service); RegistryCenter rc = registryCenterMap.get(registryCenter);
if (rc == null) {
throw Utils.runtimeExceptionWithoutStack(registryCenter + " not exists");
}
return rc.getInstance(service);
} }
public static String getServiceNameSpace(String registryCenter, String service) { public static String getServiceNameSpace(String registryCenter, String service) {

6
fizz-core/src/main/java/we/util/WebUtils.java
View File

@@ -115,10 +115,16 @@ public abstract class WebUtils {
public static final String ORIGINAL_ERROR = "origerr@"; public static final String ORIGINAL_ERROR = "origerr@";
public static final String IGNORE_PLUGIN = "ignPlg@";
private WebUtils() { private WebUtils() {
} }
public static boolean ignorePlugin(ServerWebExchange exchange) {
return exchange.getAttributes().containsKey(IGNORE_PLUGIN);
}
public static boolean isFavReq(ServerWebExchange exchange) { public static boolean isFavReq(ServerWebExchange exchange) {
return exchange.getAttribute(FAV_REQUEST) != null; return exchange.getAttribute(FAV_REQUEST) != null;
} }

2
fizz-plugin/pom.xml
View File

@@ -5,7 +5,7 @@
<parent> <parent>
<artifactId>fizz-gateway-community</artifactId> <artifactId>fizz-gateway-community</artifactId>
<groupId>com.fizzgate</groupId> <groupId>com.fizzgate</groupId>
<version>2.6.6</version> <version>2.6.7-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

495
fizz-plugin/src/main/java/we/plugin/grayrelease/GrayReleasePlugin.java Normal file
View File

@@ -0,0 +1,495 @@
/*
* Copyright (C) 2021 the original author or authors.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package we.plugin.grayrelease;
import com.auth0.jwt.JWT;
import com.auth0.jwt.interfaces.Claim;
import com.auth0.jwt.interfaces.DecodedJWT;
import com.fasterxml.jackson.core.type.TypeReference;
import inet.ipaddr.AddressStringException;
import inet.ipaddr.IPAddress;
import inet.ipaddr.IPAddressSeqRange;
import inet.ipaddr.IPAddressString;
import ognl.Ognl;
import ognl.OgnlException;
import org.apache.logging.log4j.ThreadContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.*;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.MultiValueMap;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;
import we.plugin.FizzPluginFilterChain;
import we.plugin.auth.ApiConfig;
import we.plugin.requestbody.RequestBodyPlugin;
import we.proxy.Route;
import we.spring.web.server.ext.FizzServerWebExchangeDecorator;
import we.util.*;
import java.nio.charset.StandardCharsets;
import java.util.*;
/**
* @author hongqiaowei
*/
@Component(GrayReleasePlugin.GRAY_RELEASE_PLUGIN)
public class GrayReleasePlugin extends RequestBodyPlugin {
private static final Logger LOGGER = LoggerFactory.getLogger(GrayReleasePlugin.class);
public static final String GRAY_RELEASE_PLUGIN = "GrayReleasePlugin";
private static final String triggerCondition = "triggerCondition";
private static final String routeType = "routeType";
private static final String routeConfig = "routeConfig";
private static final String routeConfigMap = "routeConfigMap";
private static final String method = "method";
private static final String path = "path";
private static final String contentType = "contentType";
private static final String body = "body";
private static final String form = "form";
private static final String cookie = "cookie";
private static final String header = "header";
private static final String query = "query";
private static final String client = "client";
private static final String ip = "ip";
private static class OgnlRoot extends HashMap<String, Object> {
public double random() {
return Math.random();
}
public boolean exist(String key) {
String[] keys = StringUtils.split(key, Consts.S.DOT);
Map<String, Object> m = this;
int keyLen = keys.length;
for (int i = 0; i < keyLen; i++) {
String k = keys[i];
if (m.containsKey(k)) {
Object obj = m.get(k);
if (obj instanceof Map) {
m = (Map<String, Object>) obj;
} else if (i + 1 != keyLen) {
return false;
}
} else {
return false;
}
}
return true;
}
public boolean matches(String key, String regex) throws OgnlException {
String value = (String) Ognl.getValue(key, this);
if (value == null) {
return false;
}
return value.matches(regex);
}
public String jwtClaim(String name) {
Map<String, Object> headerMap = (Map<String, Object>) get(GrayReleasePlugin.header);
if (headerMap == null) {
return null;
} else {
String token = (String) headerMap.get(HttpHeaders.AUTHORIZATION.toLowerCase());
if (StringUtils.isBlank(token)) {
return null;
} else if (token.length() > 7 && token.substring(0, 7).equalsIgnoreCase("Bearer ")) {
token = token.substring(7);
}
DecodedJWT jwt = JWT.decode(token);
Claim claim = jwt.getClaim(name);
if (claim == null) {
return null;
}
return claim.asString();
}
}
public boolean clientIpInRange(String range) throws AddressStringException {
Map<String, Object> cli = (Map<String, Object>) get(client);
if (cli == null) {
return false;
} else {
String pi = (String) cli.get(ip);
if (pi == null) {
return false;
} else {
return ipInRange(pi, range);
}
}
}
public boolean clientIpInRange(String rangeStartIp, String rangeEndIp) throws AddressStringException {
Map<String, Object> cli = (Map<String, Object>) get(client);
if (cli == null) {
return false;
} else {
String pi = (String) cli.get(ip);
if (pi == null) {
return false;
} else {
return ipInRange(pi, rangeStartIp, rangeEndIp);
}
}
}
public boolean ipInRange(String ip, String range) throws AddressStringException {
IPAddress ipAddress = new IPAddressString(ip).toAddress();
IPAddress rangeAddress = new IPAddressString(range).getAddress();
return rangeAddress.contains(ipAddress);
}
public boolean ipInRange(String ip, String rangeStartIp, String rangeEndIp) throws AddressStringException {
IPAddress startIPAddress = new IPAddressString(rangeStartIp).getAddress();
IPAddress endIPAddress = new IPAddressString(rangeEndIp).getAddress();
IPAddressSeqRange ipRange = startIPAddress.spanWithRange(endIPAddress);
IPAddress ipAddress = new IPAddressString(ip).toAddress();
return ipRange.contains(ipAddress);
}
public String toString() {
return JacksonUtils.writeValueAsString(this);
}
}
@Override
public Mono<Void> doFilter(ServerWebExchange exchange, Map<String, Object> config) {
String traceId = WebUtils.getTraceId(exchange);
ThreadContext.put(Consts.TRACE_ID, traceId);
String tc = (String) config.get(triggerCondition);
Object ognlRoot = request2ognlContext(exchange);
Boolean conditionMatch = false;
try {
conditionMatch = (Boolean) Ognl.getValue(tc, ognlRoot);
} catch (OgnlException e) {
LOGGER.error("calc condition expression {} with context {}", tc, ognlRoot, e);
throw new RuntimeException(e);
}
if (conditionMatch) {
Route route = WebUtils.getRoute(exchange);
changeRoute(exchange, route, config);
if (route.type == ApiConfig.Type.DIRECT_RESPONSE) {
HttpHeaders hdrs = new HttpHeaders();
hdrs.setContentType(route.contentType);
return WebUtils.response(exchange, HttpStatus.OK, hdrs, route.body);
} else {
exchange.getAttributes().put(WebUtils.IGNORE_PLUGIN, Consts.S.EMPTY);
}
}
return FizzPluginFilterChain.next(exchange);
}
private Object request2ognlContext(ServerWebExchange exchange) {
OgnlRoot ognlRoot = new OgnlRoot();
ServerHttpRequest request = exchange.getRequest();
ognlRoot.put(method, request.getMethodValue().toLowerCase());
ognlRoot.put(path, WebUtils.getClientReqPath(exchange));
MultiValueMap<String, String> queryParams = request.getQueryParams();
if (!queryParams.isEmpty()) {
Map<String, Object> queryMap = new HashMap<>();
queryParams.forEach(
(name, values) -> {
if (CollectionUtils.isEmpty(values)) {
queryMap.put(name, null);
} else if (values.size() > 1) {
queryMap.put(name, values);
} else {
queryMap.put(name, values.get(0));
}
}
);
ognlRoot.put(query, queryMap);
}
HttpHeaders headers = request.getHeaders();
if (!headers.isEmpty()) {
Map<String, Object> headerMap = new HashMap<>();
headers.forEach(
(nm, values) -> {
String name = nm.toLowerCase();
if (CollectionUtils.isEmpty(values)) {
headerMap.put(name, null);
} else if (values.size() > 1) {
headerMap.put(name, values);
} else {
headerMap.put(name, values.get(0));
}
}
);
ognlRoot.put(header, headerMap);
}
MultiValueMap<String, HttpCookie> cookies = request.getCookies();
if (!CollectionUtils.isEmpty(cookies)) {
Map<String, Object> cookieMap = new HashMap<>();
cookies.forEach(
(name, values) -> {
if (CollectionUtils.isEmpty(values)) {
cookieMap.put(name, null);
} else if (values.size() > 1) {
List<String> lst = new ArrayList<>(values.size());
for (HttpCookie value : values) {
lst.add(value.getValue());
}
cookieMap.put(name, lst);
} else {
cookieMap.put(name, values.get(0).getValue());
}
}
);
ognlRoot.put(cookie, cookieMap);
}
MediaType reqContentType = request.getHeaders().getContentType();
if (MediaType.APPLICATION_FORM_URLENCODED.isCompatibleWith(reqContentType)) {
exchange.getFormData()
.map(
formData -> {
if (formData == FizzServerWebExchangeDecorator.EMPTY_FORM_DATA) {
return null;
} else {
Map<String, Object> formMap = new HashMap<>();
formData.forEach(
(name, values) -> {
if (CollectionUtils.isEmpty(values)) {
formMap.put(name, null);
} else if (values.size() > 1) {
formMap.put(name, values);
} else {
formMap.put(name, values.get(0));
}
}
);
ognlRoot.put(form, formMap);
return formMap;
}
}
)
.subscribe();
} else if (MediaType.APPLICATION_JSON.isCompatibleWith(reqContentType)) {
request.getBody()
.single()
.map(
bodyDataBuffer -> {
if (bodyDataBuffer == NettyDataBufferUtils.EMPTY_DATA_BUFFER) {
return ReactorUtils.NULL;
} else {
String json = bodyDataBuffer.toString(StandardCharsets.UTF_8).trim();
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("request {} body: {}", request.getId(), json);
}
if (json.charAt(0) == Consts.S.LEFT_SQUARE_BRACKET) {
List<Object> bodyMap = JacksonUtils.readValue(json, new TypeReference<List<Object>>(){});
ognlRoot.put(body, bodyMap);
} else {
Map<String, Object> bodyMap = JacksonUtils.readValue(json, new TypeReference<Map<String, Object>>(){});
ognlRoot.put(body, bodyMap);
}
return ReactorUtils.NULL;
}
}
)
.subscribe();
}
String originIp = WebUtils.getOriginIp(exchange);
if (originIp != null) {
Map<String, Object> clientMap = new HashMap<>();
clientMap.put(ip, originIp);
ognlRoot.put(client, clientMap);
}
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("request {} ognl root: {}", request.getId(), JacksonUtils.writeValueAsString(ognlRoot));
}
return ognlRoot;
}
private void changeRoute(ServerWebExchange exchange, Route route, Map<String, Object> pluginConfig) {
byte rt = ((Integer) pluginConfig.get(routeType)).byteValue();
route.type = rt;
Map<String, String> newRouteConfig = (Map<String, String>) pluginConfig.get(routeConfigMap);
if (newRouteConfig == null) {
newRouteConfig = routeConfig2map((String) pluginConfig.get(routeConfig));
pluginConfig.put(routeConfigMap, newRouteConfig);
pluginConfig.remove(routeConfig);
}
if (rt == ApiConfig.Type.SERVICE_DISCOVERY) {
changeServiceDiscoveryRoute(exchange, route, newRouteConfig);
} else if (rt == ApiConfig.Type.REVERSE_PROXY) {
changeReverseProxyRoute(exchange, pluginConfig, route, newRouteConfig);
} else if (rt == ApiConfig.Type.SERVICE_AGGREGATE) {
changeAggregateRoute(exchange, route, newRouteConfig);
} else {
String ct = (String) pluginConfig.get(contentType);
String b = (String) pluginConfig.get(body);
route.contentType(MediaType.valueOf(ct))
.body(b);
}
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("route is changed: {}", route);
}
}
private void changeServiceDiscoveryRoute(ServerWebExchange exchange, Route route, Map<String, String> newRouteConfig) {
String type = newRouteConfig.get("type");
String service = newRouteConfig.get("serviceName");
if (StringUtils.isNotBlank(service)) {
route.backendService = service.trim();
}
String timeout = newRouteConfig.get("timeout");
if (StringUtils.isNotBlank(timeout)) {
route.timeout(Long.parseLong(timeout.trim()));
}
if (type.equals("http")) {
String registry = newRouteConfig.get("registry");
if (StringUtils.isNotBlank(registry)) {
route.registryCenter = registry.trim();
}
String method = newRouteConfig.get("methodName");
if (StringUtils.isNotBlank(method)) {
route.method(HttpMethod.resolve(method.trim()));
}
String path = newRouteConfig.get("path");
if (StringUtils.isNotBlank(path)) {
route.backendPath = UrlTransformUtils.transform(route.path, path.trim(), WebUtils.getClientReqPath(exchange));
}
String qry = newRouteConfig.get("query");
if (StringUtils.isNotBlank(qry)) {
route.query = qry.trim();
}
String retryCount = newRouteConfig.get("retryCount");
if (StringUtils.isNotBlank(retryCount)) {
route.retryCount(Integer.parseInt(retryCount.trim()));
}
String retryInterval = newRouteConfig.get("retryInterval");
if (StringUtils.isNotBlank(retryInterval)) {
route.retryInterval(Long.parseLong(retryInterval.trim()));
}
} else {
route.type = ApiConfig.Type.DUBBO;
String method = newRouteConfig.get("methodName");
if (StringUtils.isNotBlank(method)) {
route.rpcMethod(method.trim());
}
String version = newRouteConfig.get("version");
if (StringUtils.isNotBlank(version)) {
route.rpcVersion(version.trim());
}
String group = newRouteConfig.get("group");
if (StringUtils.isNotBlank(group)) {
route.rpcGroup(group.trim());
}
String paramTypes = newRouteConfig.get("paramTypes");
if (StringUtils.isNotBlank(paramTypes)) {
route.rpcParamTypes(paramTypes.trim());
}
}
}
private void changeReverseProxyRoute(ServerWebExchange exchange, Map<String, Object> pluginConfig, Route route, Map<String, String> newRouteConfig) {
List<String> httpHostPorts = (List<String>) pluginConfig.get("httpHostPorts");
if (httpHostPorts == null) {
String httpHostPortStr = newRouteConfig.get("serviceName");
if (StringUtils.isBlank(httpHostPortStr)) {
httpHostPorts = WebUtils.getApiConfig(exchange).httpHostPorts;
} else {
String[] httpHostPortArr = StringUtils.split(httpHostPortStr, Consts.S.COMMA);
for (int i = 0; i < httpHostPortArr.length; i++) {
httpHostPortArr[i] = httpHostPortArr[i].trim();
}
httpHostPorts = Arrays.asList(httpHostPortArr);
}
pluginConfig.put("httpHostPorts", httpHostPorts);
newRouteConfig.remove("serviceName");
}
int counter = (int) pluginConfig.getOrDefault("counter", -1);
counter++;
if (counter < 0) {
counter = Math.abs(counter);
}
String hostPort = httpHostPorts.get(
counter % httpHostPorts.size()
);
route.nextHttpHostPort(hostPort);
pluginConfig.put("counter", counter);
String method = newRouteConfig.get("methodName");
if (StringUtils.isNotBlank(method)) {
route.method(HttpMethod.resolve(method.trim()));
}
String path = newRouteConfig.get("path");
if (StringUtils.isNotBlank(path)) {
route.backendPath = UrlTransformUtils.transform(route.path, path.trim(), WebUtils.getClientReqPath(exchange));
}
String qry = newRouteConfig.get("query");
if (StringUtils.isNotBlank(qry)) {
route.query = qry.trim();
}
String timeout = newRouteConfig.get("timeout");
if (StringUtils.isNotBlank(timeout)) {
route.timeout(Long.parseLong(timeout.trim()));
}
String retryCount = newRouteConfig.get("retryCount");
if (StringUtils.isNotBlank(retryCount)) {
route.retryCount(Integer.parseInt(retryCount.trim()));
}
String retryInterval = newRouteConfig.get("retryInterval");
if (StringUtils.isNotBlank(retryInterval)) {
route.retryInterval(Long.parseLong(retryInterval.trim()));
}
}
private void changeAggregateRoute(ServerWebExchange exchange, Route route, Map<String, String> newRouteConfig) {
String service = newRouteConfig.get("serviceName");
if (StringUtils.isNotBlank(service)) {
route.backendService = service.trim();
WebUtils.setBackendService(exchange, route.backendService);
}
String path = newRouteConfig.get("path");
if (StringUtils.isNotBlank(path)) {
route.backendPath = UrlTransformUtils.transform(route.path, path.trim(), WebUtils.getClientReqPath(exchange));
WebUtils.setBackendPath(exchange, route.backendPath);
}
}
private Map<String, String> routeConfig2map(String config) {
Map<String, String> result = new HashMap<>();
String[] lines = StringUtils.split(config, Consts.S.LF);
for (String line : lines) {
int colonIdx = line.indexOf(Consts.S.COLON);
result.put(line.substring(0, colonIdx).trim(), line.substring(colonIdx + 1).trim());
}
return result;
}
}

198
fizz-plugin/src/test/java/we/plugin/grayrelease/GrayReleasePluginTests.java Normal file
View File

@@ -0,0 +1,198 @@
package we.plugin.grayrelease;
import com.fasterxml.jackson.core.type.TypeReference;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.test.web.reactive.server.WebTestClient;
import reactor.core.publisher.Mono;
import we.filter.AggregateFilter;
import we.filter.FilterResult;
import we.fizz.ConfigLoader;
import we.plugin.FizzPluginFilterChain;
import we.plugin.auth.ApiConfig;
import we.proxy.Route;
import we.util.Consts;
import we.util.JacksonUtils;
import we.util.ReflectionUtils;
import we.util.WebUtils;
import java.util.HashMap;
import java.util.Map;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
public class GrayReleasePluginTests {
/**
* service discovery backend
*/
@Test
public void simpleTest() {
final Route[] changedRoute = new Route[1];
WebTestClient client = WebTestClient.bindToWebHandler(
exchange -> {
ServerHttpResponse r = exchange.getResponse();
r.setStatusCode(HttpStatus.OK);
r.getHeaders().add(HttpHeaders.CONTENT_TYPE, MediaType.TEXT_PLAIN_VALUE);
return r.writeWith(Mono.just(r.bufferFactory().wrap("this is web handler response".getBytes())));
}
)
.webFilter(
(exchange, chain) -> {
GrayReleasePlugin grayReleasePlugin = new GrayReleasePlugin();
Map<String, Object> config = new HashMap<>();
config.put("triggerCondition", " method == 'post' " +
" and matches('path','^/apath/x*') " +
" and clientIpInRange('11.238.145.180', '11.238.145.182') " +
" and exist('body.tools.gun') ");
config.put("routeType", Integer.parseInt(String.valueOf(ApiConfig.Type.SERVICE_DISCOVERY)));
config.put("routeConfig",
"type : http \n " +
"serviceName : bservice \n " +
"path : /bpath/{$1} ");
// exchange.getAttributes().put("pcsit@", Collections.emptyIterator());
Route route = new Route().path("/apath/**");
changedRoute[0] = route;
exchange.getAttributes().put(WebUtils.ROUTE, route);
exchange.getAttributes().put(WebUtils.IGNORE_PLUGIN, Consts.S.EMPTY);
exchange.getAttributes().put(FizzPluginFilterChain.WEB_FILTER_CHAIN, chain);
exchange.getAttributes().put("oi@", "11.238.145.181");
return grayReleasePlugin.filter(exchange, config);
}
)
.build();
client.post()
.uri("/proxy/aservice/apath/xxx")
.contentType(MediaType.APPLICATION_JSON)
.bodyValue("{\"user\":\"henry\",\"tools\":{\"gun\":\"ak\"}}")
.exchange()
.expectBody(String.class).value(
v -> {
// System.err.println("body:\n" + v);
}
);
Assertions.assertEquals("bservice", changedRoute[0].backendService);
Assertions.assertEquals("/bpath/xxx", changedRoute[0].backendPath);
}
@Test
public void reverseProxyBackendTest() {
final Route[] changedRoute = new Route[1];
Map<String, Object> config = new HashMap<>();
config.put("triggerCondition", " method == 'get' ");
config.put("routeType", Integer.parseInt(String.valueOf(ApiConfig.Type.REVERSE_PROXY)));
config.put("routeConfig",
"serviceName : http://1.2.3.4:8080,http://1.2.3.5:8080 \n " +
"path : /a/b/c \n" +
"query : name1=value1&name2=value2 ");
WebTestClient client = WebTestClient.bindToWebHandler(
exchange -> {
ServerHttpResponse r = exchange.getResponse();
r.setStatusCode(HttpStatus.OK);
r.getHeaders().add(HttpHeaders.CONTENT_TYPE, MediaType.TEXT_PLAIN_VALUE);
return r.writeWith(Mono.just(r.bufferFactory().wrap("this is web handler response".getBytes())));
}
)
.webFilter(
(exchange, chain) -> {
GrayReleasePlugin grayReleasePlugin = new GrayReleasePlugin();
Route route = new Route().path("/apath/**");
changedRoute[0] = route;
exchange.getAttributes().put(WebUtils.ROUTE, route);
exchange.getAttributes().put(WebUtils.IGNORE_PLUGIN, Consts.S.EMPTY);
exchange.getAttributes().put(FizzPluginFilterChain.WEB_FILTER_CHAIN, chain);
exchange.getAttributes().put("oi@", "11.238.145.181");
return grayReleasePlugin.filter(exchange, config);
}
)
.build();
client.get()
.uri("/proxy/aservice/apath/xxx")
.exchange();
Assertions.assertEquals("/a/b/c?name1=value1&name2=value2", changedRoute[0].getBackendPathQuery());
Assertions.assertEquals("http://1.2.3.4:8080", changedRoute[0].nextHttpHostPort);
client.get()
.uri("/proxy/aservice/apath/xxx")
.exchange();
Assertions.assertEquals("http://1.2.3.5:8080", changedRoute[0].nextHttpHostPort);
client.get()
.uri("/proxy/aservice/apath/xxx")
.exchange();
Assertions.assertEquals("http://1.2.3.4:8080", changedRoute[0].nextHttpHostPort);
}
@Test
public void aggregateBackendTest() {
AggregateFilter aggregateFilter = new AggregateFilter();
ConfigLoader configLoader = mock(ConfigLoader.class);
when(
configLoader.matchAggregateResource("GET", "/_proxytest/bservice/bpath/xxx")
)
.thenReturn(null);
ReflectionUtils.set(aggregateFilter, "configLoader", configLoader);
WebTestClient client = WebTestClient.bindToWebHandler(
exchange -> {
ServerHttpResponse r = exchange.getResponse();
r.setStatusCode(HttpStatus.OK);
return r.writeWith(Mono.just(r.bufferFactory().wrap("this is web handler response".getBytes())));
}
)
.webFilter(
(exchange, chain) -> {
GrayReleasePlugin grayReleasePlugin = new GrayReleasePlugin();
Map<String, Object> config = new HashMap<>();
config.put("triggerCondition", " method == 'get' ");
config.put("routeType", Integer.parseInt(String.valueOf(ApiConfig.Type.SERVICE_AGGREGATE)));
config.put("routeConfig",
"type : http \n " +
"serviceName : bservice \n " +
"path : /bpath/{$1} ");
Route route = new Route().path("/apath/**");
exchange.getAttributes().put(WebUtils.ROUTE, route);
exchange.getAttributes().put(WebUtils.IGNORE_PLUGIN, Consts.S.EMPTY);
exchange.getAttributes().put(FizzPluginFilterChain.WEB_FILTER_CHAIN, chain);
exchange.getAttributes().put("oi@", "11.238.145.181");
Map<String, Object> filterContext = new HashMap<>();
exchange.getAttributes().put(WebUtils.FILTER_CONTEXT, filterContext);
filterContext.put(WebUtils.PREV_FILTER_RESULT, FilterResult.SUCCESS("x"));
return grayReleasePlugin.filter(exchange, config);
},
aggregateFilter
)
.build();
client.get()
.uri("/_proxytest/aservice/apath/xxx")
.exchange()
.expectBody(String.class).value(
v -> {
Map<String, Object> bodyMap = JacksonUtils.readValue(v, new TypeReference<Map<String, Object>>(){});
Assertions.assertEquals(bodyMap.get("message"), "API not found in aggregation: /_proxytest/bservice/bpath/xxx");
}
);
}
}

18
fizz-plugin/src/test/resources/log4j2-test.xml Normal file
View File

@@ -0,0 +1,18 @@
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="info">
<properties>
<property name="APP_NAME">fizz-plugin</property>
</properties>
<Appenders>
<Console name="Console" target="SYSTEM_OUT">
<PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %level %logger{36} - %msg%n"/>
</Console>
</Appenders>
<Loggers>
<Root level="warn">
<AppenderRef ref="Console"/>
</Root>
<Logger name="we" level="debug"/>
</Loggers>
</Configuration>

2
fizz-spring-boot-starter/pom.xml
View File

@@ -5,7 +5,7 @@
<parent> <parent>
<artifactId>fizz-gateway-community</artifactId> <artifactId>fizz-gateway-community</artifactId>
<groupId>com.fizzgate</groupId> <groupId>com.fizzgate</groupId>
<version>2.6.6</version> <version>2.6.7-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

18
pom.xml
View File

@@ -10,7 +10,7 @@
<reactor-bom.version>Dysprosium-SR25</reactor-bom.version> <reactor-bom.version>Dysprosium-SR25</reactor-bom.version>
<lettuce.version>5.3.7.RELEASE</lettuce.version> <lettuce.version>5.3.7.RELEASE</lettuce.version>
<nacos.cloud.version>2.2.7.RELEASE</nacos.cloud.version> <nacos.cloud.version>2.2.7.RELEASE</nacos.cloud.version>
<netty.version>4.1.78.Final</netty.version> <netty.version>4.1.79.Final</netty.version>
<httpcore.version>4.4.15</httpcore.version> <httpcore.version>4.4.15</httpcore.version>
<log4j2.version>2.17.2</log4j2.version> <log4j2.version>2.17.2</log4j2.version>
<slf4j.version>1.7.36</slf4j.version> <slf4j.version>1.7.36</slf4j.version>
@@ -22,7 +22,7 @@
<r2dbc-mysql.version>0.8.2</r2dbc-mysql.version> <r2dbc-mysql.version>0.8.2</r2dbc-mysql.version>
<reflections.version>0.9.11</reflections.version> <reflections.version>0.9.11</reflections.version>
<commons-pool2.version>2.11.1</commons-pool2.version> <commons-pool2.version>2.11.1</commons-pool2.version>
<netty-tcnative.version>2.0.53.Final</netty-tcnative.version> <netty-tcnative.version>2.0.54.Final</netty-tcnative.version>
<spring-cloud.version>2.2.9.RELEASE</spring-cloud.version> <spring-cloud.version>2.2.9.RELEASE</spring-cloud.version>
<snakeyaml.version>1.30</snakeyaml.version> <snakeyaml.version>1.30</snakeyaml.version>
<spring-data-releasetrain.version>Moore-SR13</spring-data-releasetrain.version> <spring-data-releasetrain.version>Moore-SR13</spring-data-releasetrain.version>
@@ -38,7 +38,7 @@
<artifactId>fizz-gateway-community</artifactId> <artifactId>fizz-gateway-community</artifactId>
<name>${project.artifactId}</name> <name>${project.artifactId}</name>
<description>fizz gateway community</description> <description>fizz gateway community</description>
<version>2.6.6</version> <version>2.6.7-SNAPSHOT</version>
<packaging>pom</packaging> <packaging>pom</packaging>
<modules> <modules>
<module>fizz-common</module> <module>fizz-common</module>
@@ -70,6 +70,18 @@
<dependencyManagement> <dependencyManagement>
<dependencies> <dependencies>
<dependency>
<groupId>com.github.seancfoley</groupId>
<artifactId>ipaddress</artifactId>
<version>5.3.4</version>
</dependency>
<dependency>
<groupId>ognl</groupId>
<artifactId>ognl</artifactId>
<version>3.3.3</version>
</dependency>
<dependency> <dependency>
<groupId>org.openjdk.jol</groupId> <groupId>org.openjdk.jol</groupId>
<artifactId>jol-core</artifactId> <artifactId>jol-core</artifactId>