diff --git a/src/main/java/we/FizzAppContext.java b/src/main/java/we/FizzAppContext.java new file mode 100644 index 0000000..f9647a7 --- /dev/null +++ b/src/main/java/we/FizzAppContext.java @@ -0,0 +1,9 @@ +package we; + +import org.springframework.context.ConfigurableApplicationContext; + +public class FizzAppContext { + + public static ConfigurableApplicationContext appContext; + +} diff --git a/src/main/java/we/FizzGatewayApplication.java b/src/main/java/we/FizzGatewayApplication.java index a45c023..591f6d3 100644 --- a/src/main/java/we/FizzGatewayApplication.java +++ b/src/main/java/we/FizzGatewayApplication.java @@ -1,20 +1,3 @@ -/* - * Copyright (C) 2020 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 . - */ - package we; import com.ctrip.framework.apollo.spring.annotation.EnableApolloConfig; @@ -24,20 +7,16 @@ import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration; import org.springframework.boot.autoconfigure.data.redis.RedisReactiveAutoConfiguration; import org.springframework.boot.autoconfigure.web.reactive.error.ErrorWebFluxAutoConfiguration; import org.springframework.cloud.client.discovery.EnableDiscoveryClient; -import org.springframework.context.ConfigurableApplicationContext; -/** - * @author lancer - * @author francis - */ -@SpringBootApplication(exclude = {ErrorWebFluxAutoConfiguration.class, RedisAutoConfiguration.class, RedisReactiveAutoConfiguration.class}) -// @EnableApolloConfig +@SpringBootApplication( + exclude = {ErrorWebFluxAutoConfiguration.class, RedisAutoConfiguration.class, RedisReactiveAutoConfiguration.class}, + scanBasePackages = {"we", "com.wh"} +) +@EnableApolloConfig @EnableDiscoveryClient public class FizzGatewayApplication { - public static ConfigurableApplicationContext appContext; - public static void main(String[] args) { - FizzGatewayApplication.appContext = SpringApplication.run(FizzGatewayApplication.class, args); + FizzAppContext.appContext = SpringApplication.run(FizzGatewayApplication.class, args); } } diff --git a/src/main/java/we/apollo/ApolloLog4j2ConfigurationFactory.java b/src/main/java/we/apollo/ApolloLog4j2ConfigurationFactory.java index f29889f..5c61fc9 100644 --- a/src/main/java/we/apollo/ApolloLog4j2ConfigurationFactory.java +++ b/src/main/java/we/apollo/ApolloLog4j2ConfigurationFactory.java @@ -27,8 +27,6 @@ import org.apache.logging.log4j.core.LoggerContext; import org.apache.logging.log4j.core.config.Configuration; import org.apache.logging.log4j.core.config.ConfigurationFactory; import org.apache.logging.log4j.core.config.ConfigurationSource; -import org.apache.logging.log4j.core.config.Order; -import org.apache.logging.log4j.core.config.plugins.Plugin; import org.apache.logging.log4j.core.config.properties.PropertiesConfigurationBuilder; import java.net.URI; diff --git a/src/main/java/we/config/AggrWebClientConfig.java b/src/main/java/we/config/AggrWebClientConfig.java index 6f628f7..f59742a 100644 --- a/src/main/java/we/config/AggrWebClientConfig.java +++ b/src/main/java/we/config/AggrWebClientConfig.java @@ -23,8 +23,9 @@ import org.springframework.context.annotation.Configuration; import org.springframework.web.reactive.function.client.WebClient; /** - * @author lancer + * @author hongqiaowei */ + @Configuration @ConfigurationProperties(prefix = AggrWebClientConfig.prefix) public class AggrWebClientConfig extends WebClientConfig { diff --git a/src/main/java/we/config/AppConfigProperties.java b/src/main/java/we/config/AppConfigProperties.java index f5346b6..5e19b1d 100644 --- a/src/main/java/we/config/AppConfigProperties.java +++ b/src/main/java/we/config/AppConfigProperties.java @@ -23,6 +23,7 @@ import org.springframework.context.annotation.Configuration; /** * @author unknown */ + @Configuration public class AppConfigProperties { diff --git a/src/main/java/we/config/ProxyWebClientConfig.java b/src/main/java/we/config/ProxyWebClientConfig.java index 6a4ee94..a5164c0 100644 --- a/src/main/java/we/config/ProxyWebClientConfig.java +++ b/src/main/java/we/config/ProxyWebClientConfig.java @@ -23,8 +23,9 @@ import org.springframework.context.annotation.Configuration; import org.springframework.web.reactive.function.client.WebClient; /** - * @author lancer + * @author hongqiaowei */ + @Configuration @ConfigurationProperties(prefix = ProxyWebClientConfig.prefix) public class ProxyWebClientConfig extends WebClientConfig { diff --git a/src/main/java/we/config/RedisReactiveConfig.java b/src/main/java/we/config/RedisReactiveConfig.java index a6279bc..b38424c 100644 --- a/src/main/java/we/config/RedisReactiveConfig.java +++ b/src/main/java/we/config/RedisReactiveConfig.java @@ -30,8 +30,9 @@ import org.springframework.data.redis.connection.lettuce.LettucePoolingClientCon import org.springframework.data.redis.core.ReactiveStringRedisTemplate; /** - * @author lancer + * @author hongqiaowei */ + public abstract class RedisReactiveConfig { protected static final Logger log = LoggerFactory.getLogger(RedisReactiveConfig.class); diff --git a/src/main/java/we/config/RedisReactiveProperties.java b/src/main/java/we/config/RedisReactiveProperties.java index bf40def..29cfef3 100644 --- a/src/main/java/we/config/RedisReactiveProperties.java +++ b/src/main/java/we/config/RedisReactiveProperties.java @@ -21,14 +21,15 @@ import we.util.Constants; import we.util.Utils; /** - * @author lancer + * @author hongqiaowei */ + public abstract class RedisReactiveProperties { - private String host = "127.0.0.1"; - private int port = 6379; + private String host = "127.0.0.1"; + private int port = 6379; private String password; - private int database = 0; + private int database = 0; public String getHost() { return host; diff --git a/src/main/java/we/config/SystemConfig.java b/src/main/java/we/config/SystemConfig.java index c2048d9..0d8bab6 100644 --- a/src/main/java/we/config/SystemConfig.java +++ b/src/main/java/we/config/SystemConfig.java @@ -20,11 +20,13 @@ package we.config; import com.ctrip.framework.apollo.model.ConfigChange; import com.ctrip.framework.apollo.model.ConfigChangeEvent; import com.ctrip.framework.apollo.spring.annotation.ApolloConfigChangeListener; + import we.plugin.auth.GatewayGroup; import we.util.Constants; import we.util.JacksonUtils; import we.util.NetworkUtils; import we.util.WebUtils; + import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,8 +38,9 @@ import javax.management.RuntimeErrorException; import java.util.*; /** - * @author lancer + * @author hongqiaowei */ + @Configuration public class SystemConfig { @@ -51,19 +54,19 @@ public class SystemConfig { private Set logHeaderSet = new HashSet<>(); - @Value("${gateway-group:}") - private String gatewayGroup; - - private Map> server2gatewayGroupSetMap = new HashMap<>(); - - private Set currentServerGatewayGroupSet; + // @Value("${gateway-group:}") + // private String gatewayGroup; + // + // private Map> server2gatewayGroupSetMap = new HashMap<>(); + // + // private Set currentServerGatewayGroupSet; @Value("${spring.profiles.active}") private String profile; - public Set getCurrentServerGatewayGroupSet() { - return currentServerGatewayGroupSet; - } + // public Set getCurrentServerGatewayGroupSet() { + // return currentServerGatewayGroupSet; + // } public Set getLogHeaderSet() { return logHeaderSet; @@ -73,7 +76,7 @@ public class SystemConfig { public void afterPropertiesSet() { afterLogResponseBodySet(); afterLogHeadersSet(); - afterGatewayGroupSet(); + // afterGatewayGroupSet(); } private void afterLogResponseBodySet() { @@ -90,42 +93,42 @@ public class SystemConfig { log.info("log header list: " + logHeaderSet.toString()); } - private void afterGatewayGroupSet() { - server2gatewayGroupSetMap.clear(); - if (StringUtils.isNotBlank(gatewayGroup)) { - Arrays.stream(StringUtils.split(gatewayGroup, ';')).forEach( - gg -> { - Character group = Character.valueOf(gg.charAt(0)); - String servers = gg.substring(gg.indexOf(':') + 1); - Arrays.stream(StringUtils.split(servers, ',')).forEach( - s -> { - Set gs = server2gatewayGroupSetMap.get(s); - if (gs == null) { - gs = new HashSet<>(); - server2gatewayGroupSetMap.put(s, gs); - } - gs.add(group); - } - ); - } - ); - } - log.info("server 2 gateway group set map: " + JacksonUtils.writeValueAsString(server2gatewayGroupSetMap)); - String serverIp = NetworkUtils.getServerIp(); - currentServerGatewayGroupSet = server2gatewayGroupSetMap.get(serverIp); - if (currentServerGatewayGroupSet == null) { - if (Constants.Profiles.DEV.equals(profile) || Constants.Profiles.TEST.equals(profile)) { - currentServerGatewayGroupSet = new HashSet<>(); - currentServerGatewayGroupSet.add(GatewayGroup.C); - currentServerGatewayGroupSet.add(GatewayGroup.B); - currentServerGatewayGroupSet.add(GatewayGroup.T); - server2gatewayGroupSetMap.put(serverIp, currentServerGatewayGroupSet); - } else { - throw new RuntimeException("no gateway group config for " + serverIp); - } - } - log.info("current server: " + serverIp + ", belong to: " + currentServerGatewayGroupSet); - } + // private void afterGatewayGroupSet() { + // server2gatewayGroupSetMap.clear(); + // if (StringUtils.isNotBlank(gatewayGroup)) { + // Arrays.stream(StringUtils.split(gatewayGroup, ';')).forEach( + // gg -> { + // Character group = Character.valueOf(gg.charAt(0)); + // String servers = gg.substring(gg.indexOf(':') + 1); + // Arrays.stream(StringUtils.split(servers, ',')).forEach( + // s -> { + // Set gs = server2gatewayGroupSetMap.get(s); + // if (gs == null) { + // gs = new HashSet<>(); + // server2gatewayGroupSetMap.put(s, gs); + // } + // gs.add(group); + // } + // ); + // } + // ); + // } + // log.info("server 2 gateway group set map: " + JacksonUtils.writeValueAsString(server2gatewayGroupSetMap)); + // String serverIp = NetworkUtils.getServerIp(); + // currentServerGatewayGroupSet = server2gatewayGroupSetMap.get(serverIp); + // if (currentServerGatewayGroupSet == null) { + // if (Constants.Profiles.DEV.equals(profile) || Constants.Profiles.TEST.equals(profile)) { + // currentServerGatewayGroupSet = new HashSet<>(); + // currentServerGatewayGroupSet.add(GatewayGroup.C); + // currentServerGatewayGroupSet.add(GatewayGroup.B); + // currentServerGatewayGroupSet.add(GatewayGroup.T); + // server2gatewayGroupSetMap.put(serverIp, currentServerGatewayGroupSet); + // } else { + // throw new RuntimeException("no gateway group config for " + serverIp); + // } + // } + // log.info("current server: " + serverIp + ", belong to: " + currentServerGatewayGroupSet); + // } @ApolloConfigChangeListener private void configChangeListter(ConfigChangeEvent cce) { @@ -142,10 +145,10 @@ public class SystemConfig { } else if (p.equals("log.headers")) { logHeaders = nv; afterLogHeadersSet(); - } else if (p.equals("gateway-group")) { + } /*else if (p.equals("gateway-group")) { gatewayGroup = nv; afterGatewayGroupSet(); - } + }*/ } ); } diff --git a/src/main/java/we/config/WebClientConfig.java b/src/main/java/we/config/WebClientConfig.java index 6e3b84e..21596e0 100644 --- a/src/main/java/we/config/WebClientConfig.java +++ b/src/main/java/we/config/WebClientConfig.java @@ -39,31 +39,32 @@ import java.time.Duration; import java.util.concurrent.TimeUnit; /** - * @author lancer + * @author hongqiaowei */ + public abstract class WebClientConfig { protected static final Logger log = LoggerFactory.getLogger(WebClientConfig.class); private String name; - private int maxConnections = 2_000; + private int maxConnections = 2_000; - private Duration maxIdleTime = Duration.ofMillis(40_000); + private Duration maxIdleTime = Duration.ofMillis(40_000); private Duration pendingAcquireTimeout = Duration.ofMillis(6_000); - private long connReadTimeout = 20_000; + private long connReadTimeout = 20_000; - private long connWriteTimeout = 20_000; + private long connWriteTimeout = 20_000; - private int chConnTimeout = 20_000; + private int chConnTimeout = 20_000; - private boolean chTcpNodelay = true; + private boolean chTcpNodelay = true; - private boolean chSoKeepAlive = true; + private boolean chSoKeepAlive = true; - private boolean compress = false; + private boolean compress = false; public String getName() { return name; @@ -148,7 +149,9 @@ public abstract class WebClientConfig { private ConnectionProvider getConnectionProvider() { String cpName = name + "-cp"; ConnectionProvider cp = ConnectionProvider.builder(cpName).maxConnections(maxConnections) - .pendingAcquireTimeout(pendingAcquireTimeout).maxIdleTime(maxIdleTime).build(); + .pendingAcquireTimeout(pendingAcquireTimeout) + .maxIdleTime(maxIdleTime) + .build(); log.info(cpName + ' ' + cp); return cp; } diff --git a/src/main/java/we/config/WebFluxConfig.java b/src/main/java/we/config/WebFluxConfig.java index ba7c8af..52951af 100644 --- a/src/main/java/we/config/WebFluxConfig.java +++ b/src/main/java/we/config/WebFluxConfig.java @@ -40,8 +40,9 @@ import reactor.netty.resources.LoopResources; import java.time.Duration; /** - * @author lancer + * @author hongqiaowei */ + @Configuration @ConfigurationProperties(prefix = "server") @EnableConfigurationProperties(ServerProperties.class) @@ -124,9 +125,9 @@ public class WebFluxConfig { // .channel(NioServerSocketChannel.class) .option(ChannelOption.ALLOCATOR, UnpooledByteBufAllocator.DEFAULT) // .option(ChannelOption.SO_BACKLOG, 8192) - .childOption(ChannelOption.ALLOCATOR, UnpooledByteBufAllocator.DEFAULT) + .childOption(ChannelOption.ALLOCATOR, UnpooledByteBufAllocator.DEFAULT) .childOption(ChannelOption.SO_KEEPALIVE, true) - .childOption(ChannelOption.TCP_NODELAY, true) + .childOption(ChannelOption.TCP_NODELAY, true) ) ) ); diff --git a/src/main/java/we/controller/ConfigController.java b/src/main/java/we/controller/ConfigController.java index b787340..fe0e39d 100644 --- a/src/main/java/we/controller/ConfigController.java +++ b/src/main/java/we/controller/ConfigController.java @@ -17,13 +17,13 @@ package we.controller; -import we.fizz.ConfigLoader; -import we.util.ScriptUtils; import org.apache.commons.io.FileUtils; import org.springframework.http.MediaType; import org.springframework.web.bind.annotation.*; import org.springframework.web.server.ServerWebExchange; import reactor.core.publisher.Mono; +import we.fizz.ConfigLoader; +import we.util.ScriptUtils; import javax.annotation.Resource; import java.io.File; @@ -32,6 +32,7 @@ import java.nio.charset.StandardCharsets; /** * @author unknown */ + @RestController @RequestMapping(value = "/config") public class ConfigController { @@ -45,7 +46,7 @@ public class ConfigController { return Mono.just("ok"); } - // add by lancer + // add by hongqiaowei @PostMapping(value = "/fullUpdCommonJs", consumes = MediaType.TEXT_PLAIN_VALUE) public Mono fullUpdCommonJs(ServerWebExchange exchange, @RequestBody String js) { try { diff --git a/src/main/java/we/controller/HealthController.java b/src/main/java/we/controller/HealthController.java index f9e3207..8e1c3ef 100644 --- a/src/main/java/we/controller/HealthController.java +++ b/src/main/java/we/controller/HealthController.java @@ -18,50 +18,64 @@ package we.controller; import java.util.Date; +import java.util.HashMap; +import java.util.Map; -import we.plugin.auth.ApiConfigService; -import we.plugin.auth.AppService; -import we.util.JacksonUtils; +import com.alibaba.fastjson.JSON; import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.server.ServerWebExchange; import reactor.core.publisher.Mono; +import we.plugin.auth.ApiConfigService; +import we.plugin.auth.AppService; +import we.plugin.auth.GatewayGroupService; +import we.util.JacksonUtils; import javax.annotation.Resource; /** * @author unknown */ + @RestController public class HealthController { + @Resource + private GatewayGroupService gatewayGroupService; + @Resource private AppService appService; @Resource private ApiConfigService apiConfigService; - @GetMapping("/time") - public Mono time(ServerWebExchange exchange) throws Exception{ - Date d = new Date(); - return Mono.just("Time: " + d.toString()); - } - - // add by lancer + // add by hongqiaowei @GetMapping("/sysgc") public Mono sysgc(ServerWebExchange exchange) throws Exception { System.gc(); return Mono.just("sysgc done"); } + @GetMapping("/gatewayGroups") + public Mono gatewayGroups(ServerWebExchange exchange) throws Exception { + return Mono.just(JacksonUtils.writeValueAsString(gatewayGroupService.gatewayGroupMap)); + } + + @GetMapping("/currentGatewayGroups") + public Mono currentGatewayGroups(ServerWebExchange exchange) throws Exception { + return Mono.just(JacksonUtils.writeValueAsString(gatewayGroupService.currentGatewayGroupSet)); + } + @GetMapping("/apps") public Mono apps(ServerWebExchange exchange) throws Exception { return Mono.just(JacksonUtils.writeValueAsString(appService.getAppMap())); } - @GetMapping("/apiConfigs") + @GetMapping("/serviceConfigs") public Mono apiConfigs(ServerWebExchange exchange) throws Exception { - return Mono.just(JacksonUtils.writeValueAsString(apiConfigService.getApp2gatewayGroupMap())); + return Mono.just(JacksonUtils.writeValueAsString(apiConfigService.serviceConfigMap)); } } diff --git a/src/main/java/we/controller/ManagerConfigController.java b/src/main/java/we/controller/ManagerConfigController.java index b3d83ca..92fa2f3 100644 --- a/src/main/java/we/controller/ManagerConfigController.java +++ b/src/main/java/we/controller/ManagerConfigController.java @@ -17,12 +17,6 @@ package we.controller; -import we.controller.req.BaseManagerConfigReq; -import we.controller.req.GetConfigStrReq; -import we.controller.req.GetConfigReq; -import we.controller.resp.ConfigStrResp; -import we.controller.resp.ConfigResp; -import we.fizz.ConfigLoader; import org.springframework.beans.factory.annotation.Value; import org.springframework.util.StringUtils; import org.springframework.web.bind.annotation.PostMapping; @@ -30,6 +24,12 @@ import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import reactor.core.publisher.Mono; +import we.controller.req.BaseManagerConfigReq; +import we.controller.req.GetConfigReq; +import we.controller.req.GetConfigStrReq; +import we.controller.resp.ConfigResp; +import we.controller.resp.ConfigStrResp; +import we.fizz.ConfigLoader; import javax.annotation.Resource; import javax.validation.Valid; diff --git a/src/main/java/we/controller/resp/ConfigResp.java b/src/main/java/we/controller/resp/ConfigResp.java index 49d5a84..d494175 100644 --- a/src/main/java/we/controller/resp/ConfigResp.java +++ b/src/main/java/we/controller/resp/ConfigResp.java @@ -17,12 +17,12 @@ package we.controller.resp; -import we.fizz.ConfigLoader; - import java.io.Serializable; import java.util.List; import java.util.Objects; +import we.fizz.ConfigLoader; + /** * 聚合配置响应实体类 * @author zhongjie diff --git a/src/main/java/we/filter/CorsFilterConfig.java b/src/main/java/we/filter/CorsFilterConfig.java index ebf4e2b..006839d 100644 --- a/src/main/java/we/filter/CorsFilterConfig.java +++ b/src/main/java/we/filter/CorsFilterConfig.java @@ -17,7 +17,6 @@ package we.filter; -import we.config.SystemConfig; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.Ordered; @@ -27,11 +26,14 @@ import org.springframework.web.cors.CorsConfiguration; import org.springframework.web.cors.reactive.CorsWebFilter; import org.springframework.web.cors.reactive.UrlBasedCorsConfigurationSource; +import we.config.SystemConfig; + import javax.annotation.Resource; /** - * @author lancer + * @author hongqiaowei */ + @Configuration public class CorsFilterConfig { diff --git a/src/main/java/we/filter/FilterExceptionHandlerConfig.java b/src/main/java/we/filter/FilterExceptionHandlerConfig.java index 1bdca11..b17bb1c 100644 --- a/src/main/java/we/filter/FilterExceptionHandlerConfig.java +++ b/src/main/java/we/filter/FilterExceptionHandlerConfig.java @@ -17,8 +17,6 @@ package we.filter; -import we.exception.StopAndResponseException; -import we.util.WebUtils; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.annotation.Order; @@ -29,10 +27,13 @@ import org.springframework.http.server.reactive.ServerHttpResponse; import org.springframework.web.server.ServerWebExchange; import org.springframework.web.server.WebExceptionHandler; import reactor.core.publisher.Mono; +import we.exception.StopAndResponseException; +import we.util.WebUtils; /** - * @author lancer + * @author hongqiaowei */ + @Configuration public class FilterExceptionHandlerConfig { diff --git a/src/main/java/we/filter/FilterResult.java b/src/main/java/we/filter/FilterResult.java index 67a0a75..eb6b5a9 100644 --- a/src/main/java/we/filter/FilterResult.java +++ b/src/main/java/we/filter/FilterResult.java @@ -20,8 +20,9 @@ package we.filter; import java.util.Map; /** - * @author lancer + * @author hongqiaowei */ + public class FilterResult { public String id; diff --git a/src/main/java/we/filter/FizzGatewayFilter.java b/src/main/java/we/filter/FizzGatewayFilter.java index cec27a2..27873f5 100644 --- a/src/main/java/we/filter/FizzGatewayFilter.java +++ b/src/main/java/we/filter/FizzGatewayFilter.java @@ -38,24 +38,24 @@ import org.springframework.web.server.WebFilter; import org.springframework.web.server.WebFilterChain; import com.alibaba.fastjson.JSON; -import we.flume.clients.log4j2appender.LogService; + +import io.netty.buffer.UnpooledByteBufAllocator; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; import we.constants.CommonConstants; import we.fizz.AggregateResource; import we.fizz.AggregateResult; import we.fizz.ConfigLoader; import we.fizz.Pipeline; import we.fizz.input.Input; +import we.flume.clients.log4j2appender.LogService; import we.util.Constants; import we.util.MapUtil; import we.util.WebUtils; -import io.netty.buffer.UnpooledByteBufAllocator; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.core.scheduler.Schedulers; - /** - * @author lancer + * @author francis */ @Component @Order(2) diff --git a/src/main/java/we/filter/FizzLogFilter.java b/src/main/java/we/filter/FizzLogFilter.java index 6653ba5..3a1f054 100644 --- a/src/main/java/we/filter/FizzLogFilter.java +++ b/src/main/java/we/filter/FizzLogFilter.java @@ -17,8 +17,6 @@ package we.filter; -import we.flume.clients.log4j2appender.LogService; -import we.util.WebUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.core.annotation.Order; @@ -27,10 +25,14 @@ import org.springframework.web.server.ServerWebExchange; import org.springframework.web.server.WebFilter; import org.springframework.web.server.WebFilterChain; import reactor.core.publisher.Mono; +import we.flume.clients.log4j2appender.LogService; +import we.util.ThreadContext; +import we.util.WebUtils; /** - * @author lancer + * @author hongqiaowei */ + @Component @Order(0) public class FizzLogFilter implements WebFilter { @@ -46,7 +48,7 @@ public class FizzLogFilter implements WebFilter { long startTime = System.currentTimeMillis(); return chain.filter(exchange).doAfterTerminate( () -> { - StringBuilder b = new StringBuilder(160); + StringBuilder b = ThreadContext.getStringBuilder(); WebUtils.request2stringBuilder(exchange, b); b.append(resp).append(exchange.getResponse().getStatusCode()) .append(in) .append(System.currentTimeMillis() - startTime); diff --git a/src/main/java/we/filter/PreFilter.java b/src/main/java/we/filter/PreFilter.java index 01d2615..1e2241e 100644 --- a/src/main/java/we/filter/PreFilter.java +++ b/src/main/java/we/filter/PreFilter.java @@ -17,15 +17,6 @@ package we.filter; -import we.config.SystemConfig; -import we.plugin.FixedPluginFilter; -import we.plugin.PluginConfig; -import we.plugin.PluginFilter; -import we.plugin.auth.*; -import we.plugin.stat.StatPluginFilter; -import we.util.Constants; -import we.util.ReactorUtils; -import we.util.WebUtils; import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -37,15 +28,24 @@ import org.springframework.stereotype.Component; import org.springframework.web.server.ServerWebExchange; import org.springframework.web.server.WebFilterChain; import reactor.core.publisher.Mono; +import we.plugin.FixedPluginFilter; +import we.plugin.PluginConfig; +import we.plugin.PluginFilter; +import we.plugin.auth.ApiConfig; +import we.plugin.auth.ApiConfigService; +import we.plugin.auth.AuthPluginFilter; +import we.plugin.stat.StatPluginFilter; +import we.util.ReactorUtils; +import we.util.WebUtils; -import javax.annotation.PostConstruct; import javax.annotation.Resource; import java.util.*; import java.util.function.Function; /** - * @author lancer + * @author hongqiaowei */ + @Component(PreFilter.PRE_FILTER) @Order(1) public class PreFilter extends ProxyAggrFilter { @@ -62,25 +62,12 @@ public class PreFilter extends ProxyAggrFilter { @Value("${b-services:x}") private Set bServices = new HashSet<>(); - @Resource - private SystemConfig systemConfig; - @Resource(name = StatPluginFilter.STAT_PLUGIN_FILTER) private StatPluginFilter statPluginFilter; @Resource(name = AuthPluginFilter.AUTH_PLUGIN_FILTER) private AuthPluginFilter authPluginFilter; - private char currentGatewayGroup; - @PostConstruct - public void setCurrentGatewayGroup() { - for (Character gg : systemConfig.getCurrentServerGatewayGroupSet()) { - currentGatewayGroup = gg.charValue(); - log.info("current gateway group is " + currentGatewayGroup); - break; - } - } - @Override public Mono doFilter(ServerWebExchange exchange, WebFilterChain chain) { @@ -88,24 +75,11 @@ public class PreFilter extends ProxyAggrFilter { Map appendHdrs = new HashMap<>(6, 1.0f); Map eas = exchange.getAttributes(); eas.put(WebUtils.FILTER_CONTEXT, fc); eas.put(WebUtils.APPEND_HEADERS, appendHdrs); - eas.put(WebUtils.CGG, currentGatewayGroup); String app = WebUtils.getHeaderValue(exchange, WebUtils.APP_HEADER); - if (StringUtils.isBlank(app)) { - if (Constants.Profiles.DEV.equals(profile) || Constants.Profiles.TEST.equals(profile)) { - String service = WebUtils.getServiceId(exchange); - if (bServices.contains(service)) { - app = App.TO_B; - } else { - app = App.TO_C; - } - } else if (currentGatewayGroup == GatewayGroup.B) { - app = App.TO_B; - } else { - app = App.TO_C; - } + if (StringUtils.isNotBlank(app)) { + eas.put(WebUtils.APP_HEADER, app); } - eas.put(WebUtils.APP_HEADER, app); Mono vm = statPluginFilter.filter(exchange, null, null); return chain(exchange, vm, authPluginFilter).defaultIfEmpty(ReactorUtils.NULL) @@ -120,11 +94,7 @@ public class PreFilter extends ProxyAggrFilter { if (ac.pluginConfigs == null || ac.pluginConfigs.isEmpty()) { return m.flatMap(func(exchange, chain)); } else { - return m.flatMap( - e -> { - return executeManagedPluginFilters(exchange, ac.pluginConfigs); - } - ) + return m.flatMap(e -> {return executeManagedPluginFilters(exchange, ac.pluginConfigs);}) .defaultIfEmpty(ReactorUtils.NULL).flatMap(func(exchange, chain)); } } else if (authRes == ApiConfigService.Access.YES) { diff --git a/src/main/java/we/filter/ProxyAggrFilter.java b/src/main/java/we/filter/ProxyAggrFilter.java index a7a90c1..99529ca 100644 --- a/src/main/java/we/filter/ProxyAggrFilter.java +++ b/src/main/java/we/filter/ProxyAggrFilter.java @@ -17,15 +17,16 @@ package we.filter; -import we.util.WebUtils; import org.springframework.web.server.ServerWebExchange; import org.springframework.web.server.WebFilter; import org.springframework.web.server.WebFilterChain; import reactor.core.publisher.Mono; +import we.util.WebUtils; /** - * @author lancer + * @author hongqiaowei */ + // TODO 类名调整 public abstract class ProxyAggrFilter implements WebFilter { diff --git a/src/main/java/we/filter/RouteFilter.java b/src/main/java/we/filter/RouteFilter.java index 7fe853e..1c8581f 100644 --- a/src/main/java/we/filter/RouteFilter.java +++ b/src/main/java/we/filter/RouteFilter.java @@ -17,11 +17,6 @@ package we.filter; -import we.flume.clients.log4j2appender.LogService; -import we.legacy.RespEntity; -import we.proxy.FizzWebClient; -import we.util.ThreadContext; -import we.util.WebUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.core.Ordered; @@ -36,14 +31,23 @@ import org.springframework.web.reactive.function.client.ClientResponse; import org.springframework.web.server.ServerWebExchange; import org.springframework.web.server.WebFilterChain; import reactor.core.publisher.Mono; +import we.flume.clients.log4j2appender.LogService; +import we.legacy.RespEntity; +import we.plugin.auth.ApiConfig; +import we.plugin.auth.AuthPluginFilter; +import we.proxy.FizzWebClient; +import we.util.ThreadContext; +import we.util.WebUtils; import javax.annotation.Resource; import java.util.List; import java.util.Map; +import java.util.function.Function; /** - * @author lancer + * @author hongqiaowei */ + @Component @Order(Ordered.LOWEST_PRECEDENCE) public class RouteFilter extends ProxyAggrFilter { @@ -79,7 +83,6 @@ public class RouteFilter extends ProxyAggrFilter { private Mono doFilter0(ServerWebExchange exchange, WebFilterChain chain) { ServerHttpRequest clientReq = exchange.getRequest(); - String rid = clientReq.getId(); HttpHeaders hdrs = new HttpHeaders(); clientReq.getHeaders().forEach( (h, vs) -> { @@ -101,37 +104,65 @@ public class RouteFilter extends ProxyAggrFilter { ); } - return fizzWebClient.proxySend2service(rid, clientReq.getMethod(), WebUtils.getServiceId(exchange), WebUtils.getRelativeUri(exchange), hdrs, exchange.getRequest().getBody()).flatMap( - remoteResp -> { - ServerHttpResponse clientResp = exchange.getResponse(); - clientResp.setStatusCode(remoteResp.statusCode()); - HttpHeaders clientRespHeaders = clientResp.getHeaders(); - HttpHeaders remoteRespHeaders = remoteResp.headers().asHttpHeaders(); - remoteRespHeaders.entrySet().forEach( - h -> { - String k = h.getKey(); - if (clientRespHeaders.containsKey(k)) { - if (k.equals(HttpHeaders.ACCESS_CONTROL_ALLOW_ORIGIN) || k.equals(HttpHeaders.ACCESS_CONTROL_ALLOW_CREDENTIALS) - || k.equals(HttpHeaders.ACCESS_CONTROL_ALLOW_HEADERS) || k.equals(HttpHeaders.ACCESS_CONTROL_MAX_AGE)) { - } else { - clientRespHeaders.put(k, h.getValue()); - } - } else { - clientRespHeaders.put(k, h.getValue()); - } - } - ); - if (log.isDebugEnabled()) { - StringBuilder b = ThreadContext.getStringBuilder(); - WebUtils.response2stringBuilder(rid, remoteResp, b); - log.debug(b.toString(), LogService.BIZ_ID, rid); - } - return clientResp.writeWith(remoteResp.body(BodyExtractors.toDataBuffers())) - .doOnError(throwable -> cleanup(remoteResp)).doOnCancel(() -> cleanup(remoteResp)); - } - ); + ApiConfig ac = null; + Object authRes = WebUtils.getFilterResultDataItem(exchange, AuthPluginFilter.AUTH_PLUGIN_FILTER, AuthPluginFilter.RESULT); + if (authRes instanceof ApiConfig) { + ac = (ApiConfig) authRes; + } + + String relativeUri = WebUtils.getRelativeUri(exchange); + if (ac == null || ac.proxyMode == ApiConfig.DIRECT_PROXY_MODE) { + return send(exchange, WebUtils.getServiceId(exchange), relativeUri, hdrs); + } else { + String realUri; + String backendUrl = ac.getNextBackendUrl(); + int acpLen = ac.path.length(); + if (acpLen == 1) { + realUri = backendUrl + relativeUri; + } else { + realUri = backendUrl + relativeUri.substring(acpLen); + } + relativeUri.substring(acpLen); + return fizzWebClient.send(clientReq.getId(), clientReq.getMethod(), realUri, hdrs, clientReq.getBody()).flatMap(genServerResponse(exchange)); + } } - + + private Mono send(ServerWebExchange exchange, String service, String relativeUri, HttpHeaders hdrs) { + ServerHttpRequest clientReq = exchange.getRequest(); + return fizzWebClient.proxySend2service(clientReq.getId(), clientReq.getMethod(), service, relativeUri, hdrs, clientReq.getBody()).flatMap(genServerResponse(exchange)); + } + + private Function> genServerResponse(ServerWebExchange exchange) { + return remoteResp -> { + ServerHttpResponse clientResp = exchange.getResponse(); + clientResp.setStatusCode(remoteResp.statusCode()); + HttpHeaders clientRespHeaders = clientResp.getHeaders(); + HttpHeaders remoteRespHeaders = remoteResp.headers().asHttpHeaders(); + remoteRespHeaders.entrySet().forEach( + h -> { + String k = h.getKey(); + if (clientRespHeaders.containsKey(k)) { + if (k.equals(HttpHeaders.ACCESS_CONTROL_ALLOW_ORIGIN) || k.equals(HttpHeaders.ACCESS_CONTROL_ALLOW_CREDENTIALS) + || k.equals(HttpHeaders.ACCESS_CONTROL_ALLOW_HEADERS) || k.equals(HttpHeaders.ACCESS_CONTROL_MAX_AGE)) { + } else { + clientRespHeaders.put(k, h.getValue()); + } + } else { + clientRespHeaders.put(k, h.getValue()); + } + } + ); + if (log.isDebugEnabled()) { + StringBuilder b = ThreadContext.getStringBuilder(); + String rid = exchange.getRequest().getId(); + WebUtils.response2stringBuilder(rid, remoteResp, b); + log.debug(b.toString(), LogService.BIZ_ID, rid); + } + return clientResp.writeWith(remoteResp.body(BodyExtractors.toDataBuffers())) + .doOnError(throwable -> cleanup(remoteResp)).doOnCancel(() -> cleanup(remoteResp)); + }; + } + private void cleanup(ClientResponse clientResponse) { if (clientResponse != null) { clientResponse.bodyToMono(Void.class).subscribe(); diff --git a/src/main/java/we/fizz/ConfigLoader.java b/src/main/java/we/fizz/ConfigLoader.java index df7146f..e8a661c 100644 --- a/src/main/java/we/fizz/ConfigLoader.java +++ b/src/main/java/we/fizz/ConfigLoader.java @@ -19,10 +19,12 @@ package we.fizz; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONArray; + import we.config.AppConfigProperties; import we.fizz.input.ClientInputConfig; import we.fizz.input.Input; import we.fizz.input.InputType; + import org.apache.commons.io.FileUtils; import org.noear.snack.ONode; import org.springframework.beans.factory.annotation.Value; @@ -32,6 +34,10 @@ import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; import javax.annotation.Resource; + +import static we.listener.AggregateRedisConfig.AGGREGATE_REACTIVE_REDIS_TEMPLATE; +import static we.util.Constants.Symbol.FORWARD_SLASH; + import java.io.File; import java.io.IOException; import java.io.Serializable; @@ -41,9 +47,6 @@ import java.util.List; import java.util.Map; import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; - -import static we.listener.AggregateRedisConfig.AGGREGATE_REACTIVE_REDIS_TEMPLATE; -import static we.util.Constants.Symbol.FORWARD_SLASH; /** * * @author francis @@ -56,7 +59,7 @@ public class ConfigLoader { * 聚合配置存放Hash的Key */ private static final String AGGREGATE_HASH_KEY = "fizz_aggregate_config"; - + private static Map aggregateResources = null; private static Map resourceKey2ConfigInfoMap = null; private static Map aggregateId2ResourceKeyMap = null; @@ -80,8 +83,13 @@ public class ConfigLoader { clientInputConfig.setHeaders(cfgNode.select("$.headers").toObject(Map.class)); clientInputConfig.setMethod(cfgNode.select("$.method").getString()); clientInputConfig.setPath(cfgNode.select("$.path").getString()); - if(cfgNode.select("$.debug") != null) { - clientInputConfig.setDebug(cfgNode.select("$.debug").getBoolean()); + if(clientInputConfig.getPath().startsWith(TEST_PATH_PREFIX)) { + // always enable debug for testing + clientInputConfig.setDebug(true); + }else { + if(cfgNode.select("$.debug") != null) { + clientInputConfig.setDebug(cfgNode.select("$.debug").getBoolean()); + } } clientInputConfig.setType(InputType.valueOf(cfgNode.select("$.type").getString())); clientInputConfig.setLangDef(cfgNode.select("$.langDef").toObject(Map.class)); diff --git a/src/main/java/we/fizz/Pipeline.java b/src/main/java/we/fizz/Pipeline.java index 9b5d8ba..87baf7c 100644 --- a/src/main/java/we/fizz/Pipeline.java +++ b/src/main/java/we/fizz/Pipeline.java @@ -25,11 +25,7 @@ import java.util.Map; import javax.script.ScriptException; -import we.fizz.input.ClientInputConfig; -import we.fizz.input.InputConfig; import we.schema.util.I18nUtils; -import we.util.JsonSchemaUtils; - import org.noear.snack.ONode; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -37,14 +33,17 @@ import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; import com.alibaba.fastjson.JSON; -import we.flume.clients.log4j2appender.LogService; -import we.fizz.input.Input; -import we.fizz.input.PathMapping; -import we.fizz.input.ScriptHelper; -import we.util.MapUtil; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import we.fizz.input.ClientInputConfig; +import we.fizz.input.Input; +import we.fizz.input.InputConfig; +import we.fizz.input.PathMapping; +import we.fizz.input.ScriptHelper; +import we.flume.clients.log4j2appender.LogService; +import we.util.JsonSchemaUtils; +import we.util.MapUtil; /** * diff --git a/src/main/java/we/fizz/Step.java b/src/main/java/we/fizz/Step.java index 9a44f41..93d0bd1 100644 --- a/src/main/java/we/fizz/Step.java +++ b/src/main/java/we/fizz/Step.java @@ -30,14 +30,14 @@ import org.springframework.web.reactive.function.BodyInserters; import org.springframework.web.reactive.function.client.WebClient; import com.alibaba.fastjson.JSON; + +import reactor.core.publisher.Mono; import we.fizz.input.Input; import we.fizz.input.InputConfig; import we.fizz.input.InputContext; import we.fizz.input.InputFactory; import we.fizz.input.InputType; -import reactor.core.publisher.Mono; - /** * * @author linwaiwai diff --git a/src/main/java/we/fizz/input/Input.java b/src/main/java/we/fizz/input/Input.java index 9cd571c..0904b68 100644 --- a/src/main/java/we/fizz/input/Input.java +++ b/src/main/java/we/fizz/input/Input.java @@ -19,11 +19,10 @@ package we.fizz.input; import java.util.HashMap; import java.util.Map; +import reactor.core.publisher.Mono; import we.fizz.StepContext; import we.fizz.StepResponse; -import reactor.core.publisher.Mono; - /** * * @author linwaiwai diff --git a/src/main/java/we/fizz/input/RequestInput.java b/src/main/java/we/fizz/input/RequestInput.java index e5ebe79..0494390 100644 --- a/src/main/java/we/fizz/input/RequestInput.java +++ b/src/main/java/we/fizz/input/RequestInput.java @@ -34,16 +34,16 @@ import org.springframework.web.util.UriComponents; import org.springframework.web.util.UriComponentsBuilder; import com.alibaba.fastjson.JSON; -import we.flume.clients.log4j2appender.LogService; -import we.FizzGatewayApplication; + +import reactor.core.publisher.Mono; +import we.FizzAppContext; import we.constants.CommonConstants; import we.fizz.StepContext; import we.fizz.StepResponse; +import we.flume.clients.log4j2appender.LogService; import we.proxy.FizzWebClient; import we.util.MapUtil; -import reactor.core.publisher.Mono; - /** * * @author linwaiwai @@ -222,7 +222,7 @@ public class RequestInput extends Input { String aggrPath = (String)inputContext.getStepContext().getInputReqAttr("path"); String aggrService = aggrPath.split("\\/")[2]; - FizzWebClient client = FizzGatewayApplication.appContext.getBean(FizzWebClient.class); + FizzWebClient client = FizzAppContext.appContext.getBean(FizzWebClient.class); return client.aggrSend(aggrService, aggrMethod, aggrPath, null, method, url, MapUtil.toHttpHeaders(headers), request.get("body"), (long)timeout); } diff --git a/src/main/java/we/fizz/input/ScriptHelper.java b/src/main/java/we/fizz/input/ScriptHelper.java index 15e42cb..a6644d9 100644 --- a/src/main/java/we/fizz/input/ScriptHelper.java +++ b/src/main/java/we/fizz/input/ScriptHelper.java @@ -30,6 +30,7 @@ import org.slf4j.LoggerFactory; import com.alibaba.fastjson.JSON; import com.ctrip.framework.apollo.core.utils.StringUtils; + import we.constants.CommonConstants; import we.exception.StopAndResponseException; import we.fizz.StepContext; diff --git a/src/main/java/we/flume/clients/log4j2appender/LogService.java b/src/main/java/we/flume/clients/log4j2appender/LogService.java index 0aaf5cf..5ae538d 100644 --- a/src/main/java/we/flume/clients/log4j2appender/LogService.java +++ b/src/main/java/we/flume/clients/log4j2appender/LogService.java @@ -1,33 +1,31 @@ -/* - * Copyright (C) 2020 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 . - */ - package we.flume.clients.log4j2appender; -/** - * @author lancer - */ public enum LogService { - BIZ_ID, HANDLE_STGY; + BIZ_ID, HANDLE_STGY, APP; + + public static void cleanBizId() { + setBizId(null); + } + + public static Object getBizId() { + return ThreadContext.get(Constants.BIZ_ID); + } public static void setBizId(Object bizId) { + ThreadContext.set(Constants.BIZ_ID, bizId); } public static String toKF(String topic) { return topic; } + + public static String toESaKF(String topic) { + return Constants.AND + topic; + } + + public static class Constants { + static final String BIZ_ID = "bizId"; + static final char AND = '&'; + } } diff --git a/src/main/java/we/flume/clients/log4j2appender/ThreadContext.java b/src/main/java/we/flume/clients/log4j2appender/ThreadContext.java new file mode 100644 index 0000000..856d58d --- /dev/null +++ b/src/main/java/we/flume/clients/log4j2appender/ThreadContext.java @@ -0,0 +1,76 @@ +package we.flume.clients.log4j2appender; + +import java.text.SimpleDateFormat; +import java.util.HashMap; +import java.util.Map; + +/** for internal use */ +public abstract class ThreadContext { + + private static ThreadLocal> tl = new ThreadLocal<>(); + private static final int mapCap = 16; + private static final float mapLoadFactor = 1.0f; + + private static final String sb = "sb"; + private static final int sbCap = 256; + + public static StringBuilder getStringBuilder() { + return getStringBuilder(true); + } + + public static StringBuilder getStringBuilder(boolean clean) { + Map m = getMap(); + StringBuilder b = (StringBuilder) m.get(sb); + if (b == null) { + b = new StringBuilder(sbCap); + m.put(sb, b); + } else { + if (clean) { + b.delete(0, b.length()); + } + } + return b; + } + + public static SimpleDateFormat getSimpleDateFormat(String pattern) { + Map m = getMap(); + SimpleDateFormat sdf = (SimpleDateFormat) m.get(pattern); + if (sdf == null) { + sdf = new SimpleDateFormat(pattern); + m.put(pattern, sdf); + } + return sdf; + } + + public static Object get(String key, Class clz) { + Object obj = get(key); + if (obj == null) { + try { + obj = clz.newInstance(); + set(key, obj); + } catch (InstantiationException e) { + throw new RuntimeException(e); + } catch (IllegalAccessException e) { + throw new RuntimeException(e); + } + } + return obj; + } + + private static Map getMap() { + Map m = tl.get(); + if (m == null) { + m = new HashMap<>(mapCap, mapLoadFactor); + tl.set(m); + } + return m; + } + + public static Object get(String key) { + return getMap().get(key); + } + + public static void set(String key, Object obj) { + getMap().put(key, obj); + } +} diff --git a/src/main/java/we/legacy/RespEntity.java b/src/main/java/we/legacy/RespEntity.java index b3a656d..4a81034 100644 --- a/src/main/java/we/legacy/RespEntity.java +++ b/src/main/java/we/legacy/RespEntity.java @@ -17,13 +17,15 @@ package we.legacy; -import we.util.Constants; -import we.util.ThreadContext; import org.springframework.lang.Nullable; +import we.util.Constants; +import we.util.ThreadContext; + /** - * @author lancer + * @author hongqiaowei */ + public class RespEntity { private static final String f0 = "{\"msgCode\":"; diff --git a/src/main/java/we/listener/AggregateChannelListener.java b/src/main/java/we/listener/AggregateChannelListener.java index 24374d9..77c3db7 100644 --- a/src/main/java/we/listener/AggregateChannelListener.java +++ b/src/main/java/we/listener/AggregateChannelListener.java @@ -19,7 +19,7 @@ package we.listener; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.TypeReference; -import we.fizz.ConfigLoader; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Qualifier; @@ -31,14 +31,16 @@ import org.springframework.util.StringUtils; import reactor.core.Disposable; import reactor.core.publisher.Flux; import reactor.core.scheduler.Schedulers; +import we.fizz.ConfigLoader; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; -import java.net.InetAddress; -import java.util.List; import static we.listener.AggregateRedisConfig.AGGREGATE_REACTIVE_REDIS_MESSAGE_LISTENER_CONTAINER; +import java.net.InetAddress; +import java.util.List; + /** * 聚合Channel监听器 * @author zhongjie diff --git a/src/main/java/we/listener/AggregateRedisConfig.java b/src/main/java/we/listener/AggregateRedisConfig.java index 252025d..86c7d31 100644 --- a/src/main/java/we/listener/AggregateRedisConfig.java +++ b/src/main/java/we/listener/AggregateRedisConfig.java @@ -17,8 +17,6 @@ package we.listener; -import we.config.RedisReactiveConfig; -import we.config.RedisReactiveProperties; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Bean; @@ -27,6 +25,9 @@ import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory; import org.springframework.data.redis.core.ReactiveStringRedisTemplate; import org.springframework.data.redis.listener.ReactiveRedisMessageListenerContainer; +import we.config.RedisReactiveConfig; +import we.config.RedisReactiveProperties; + /** * 聚合配置Redis配置 * @author zhongjie diff --git a/src/main/java/we/plugin/FixedPluginFilter.java b/src/main/java/we/plugin/FixedPluginFilter.java index d56c5ff..02ea426 100644 --- a/src/main/java/we/plugin/FixedPluginFilter.java +++ b/src/main/java/we/plugin/FixedPluginFilter.java @@ -14,16 +14,19 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.plugin; -import we.FizzGatewayApplication; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import we.FizzAppContext; +import we.util.ThreadContext; + import java.util.*; /** - * @author lancer + * @author hongqiaowei */ public abstract class FixedPluginFilter extends PluginFilter { @@ -56,7 +59,7 @@ public abstract class FixedPluginFilter extends PluginFilter { ); String fid = pf.getId(); fixedPluginFilterMap.put(fid, pf); - StringBuilder b = new StringBuilder(256); + StringBuilder b = ThreadContext.getStringBuilder(); b.append("add ").append(fid).append('\n'); b.append("fixed plugin filters: \n"); filters2sb(b); @@ -67,7 +70,7 @@ public abstract class FixedPluginFilter extends PluginFilter { if (fixedPluginFilterList == null) { synchronized (fixedPluginFilterMap) { if (fixedPluginFilterList == null) { - Map beansOfType = FizzGatewayApplication.appContext.getBeansOfType(FixedPluginFilter.class); + Map beansOfType = FizzAppContext.appContext.getBeansOfType(FixedPluginFilter.class); if (beansOfType == null || beansOfType.isEmpty()) { fixedPluginFilterList = Collections.EMPTY_LIST; } else { @@ -82,7 +85,7 @@ public abstract class FixedPluginFilter extends PluginFilter { fixedPluginFilterMap.put(f.getId(), f); } ); - StringBuilder b = new StringBuilder(256); + StringBuilder b = ThreadContext.getStringBuilder(); b.append("fixed plugin filters: \n"); filters2sb(b); log.info(b.toString()); diff --git a/src/main/java/we/plugin/PluginConfig.java b/src/main/java/we/plugin/PluginConfig.java index 1fcd1fa..1b8b1e8 100644 --- a/src/main/java/we/plugin/PluginConfig.java +++ b/src/main/java/we/plugin/PluginConfig.java @@ -14,17 +14,17 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.plugin; -import com.fasterxml.jackson.core.JsonProcessingException; -import we.util.JacksonUtils; import org.apache.commons.lang3.StringUtils; +import we.util.JacksonUtils; import java.util.Collections; import java.util.Map; /** - * @author lancer + * @author hongqiaowei */ public class PluginConfig { diff --git a/src/main/java/we/plugin/PluginFilter.java b/src/main/java/we/plugin/PluginFilter.java index 41edf4a..0de82b6 100644 --- a/src/main/java/we/plugin/PluginFilter.java +++ b/src/main/java/we/plugin/PluginFilter.java @@ -14,23 +14,24 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.plugin; -import we.flume.clients.log4j2appender.LogService; -import we.filter.FilterResult; -import we.legacy.RespEntity; -import we.util.WebUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.http.HttpStatus; import org.springframework.http.server.reactive.ServerHttpRequest; import org.springframework.web.server.ServerWebExchange; import reactor.core.publisher.Mono; +import we.filter.FilterResult; +import we.flume.clients.log4j2appender.LogService; +import we.legacy.RespEntity; +import we.util.WebUtils; import java.util.Map; /** - * @author lancer + * @author hongqiaowei */ public abstract class PluginFilter { diff --git a/src/main/java/we/plugin/auth/ApiConfig.java b/src/main/java/we/plugin/auth/ApiConfig.java index 03ea86f..99e7281 100644 --- a/src/main/java/we/plugin/auth/ApiConfig.java +++ b/src/main/java/we/plugin/auth/ApiConfig.java @@ -14,66 +14,90 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.plugin.auth; +import com.fasterxml.jackson.annotation.JsonIgnore; +import org.apache.commons.lang3.StringUtils; +import org.springframework.http.HttpMethod; + import we.plugin.PluginConfig; import we.util.Constants; import we.util.JacksonUtils; -import org.apache.commons.lang3.StringUtils; -import org.springframework.http.HttpMethod; -import java.util.List; +import java.util.*; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; /** - * @author lancer + * @author hongqiaowei */ public class ApiConfig { - public static final int DELETED = 1; + public static final int DELETED = 1; - public static final char ALLOW = 'a'; + public static final char ALLOW = 'a'; - public static final char FORBID = 'f'; + public static final char FORBID = 'f'; - public int id; // tb_api_auth.id + public static final byte DIRECT_PROXY_MODE = 1; - public int isDeleted = 0; // tb_api_auth.is_deleted + public static final byte PREFIX_REWRITE_PROXY_MODE = 2; - public char gatewayGroup; // tb_api_auth.gateway_group + @JsonIgnore + public int id; // tb_api_auth.id + + @JsonIgnore + public int isDeleted = 0; // tb_api_auth.is_deleted + + public Set gatewayGroups = new HashSet<>(6); public String service; - public HttpMethod method = HttpMethod.X; + public HttpMethod method = HttpMethod.X; - public String path = String.valueOf(Constants.Symbol.FORWARD_SLASH); + public String path = String.valueOf(Constants.Symbol.FORWARD_SLASH); - private String app; + public Set apps = new HashSet<>(6); - public char access = ALLOW; + public byte proxyMode = DIRECT_PROXY_MODE; + + private AtomicInteger counter = new AtomicInteger(-1); + + public List backendUrls; + + public char access = ALLOW; public List pluginConfigs; - public void setApp(String a) { - app = a; + public void setGatewayGroup(String ggs) { + if (StringUtils.isBlank(ggs)) { + gatewayGroups.add("*"); + } else { + Arrays.stream(StringUtils.split(ggs, ',')).forEach( + gg -> { + gatewayGroups.add(gg.trim()); + } + ); + } } - public String app() { - if (StringUtils.isBlank(app)) { - if (gatewayGroup == GatewayGroup.C) { - app = App.TO_C; - } else if (gatewayGroup == GatewayGroup.B) { - app = App.TO_B; - } else { - throw new RuntimeException(toString() + " no app", null, false, false) {}; - } + public void setApp(String as) { + if (StringUtils.isBlank(as)) { + apps.add("*"); + } else { + Arrays.stream(StringUtils.split(as, ',')).forEach( + a -> { + apps.add(a.trim()); + } + ); } - return app; } public void setPath(String p) { if (StringUtils.isNotBlank(p)) { - path = p; + path = p.trim(); } } @@ -84,6 +108,16 @@ public class ApiConfig { } } + @JsonIgnore + public String getNextBackendUrl() { + int idx = counter.incrementAndGet(); + if (idx < 0) { + counter.set(0); + idx = 0; + } + return backendUrls.get(idx % backendUrls.size()); + } + @Override public String toString() { return JacksonUtils.writeValueAsString(this); diff --git a/src/main/java/we/plugin/auth/ApiConfigService.java b/src/main/java/we/plugin/auth/ApiConfigService.java index 83c5dc9..8c1737a 100644 --- a/src/main/java/we/plugin/auth/ApiConfigService.java +++ b/src/main/java/we/plugin/auth/ApiConfigService.java @@ -14,15 +14,12 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.plugin.auth; import com.ctrip.framework.apollo.model.ConfigChange; import com.ctrip.framework.apollo.model.ConfigChangeEvent; import com.ctrip.framework.apollo.spring.annotation.ApolloConfigChangeListener; -import we.flume.clients.log4j2appender.LogService; -import we.config.SystemConfig; -import we.listener.AggregateRedisConfig; -import we.util.*; import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,6 +33,9 @@ import org.springframework.stereotype.Service; import org.springframework.web.server.ServerWebExchange; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import we.flume.clients.log4j2appender.LogService; +import we.listener.AggregateRedisConfig; +import we.util.*; import javax.annotation.PostConstruct; import javax.annotation.Resource; @@ -43,7 +43,7 @@ import java.util.*; import java.util.concurrent.TimeUnit; /** - * @author lancer + * @author hongqiaowei */ @Service @@ -61,9 +61,9 @@ public class ApiConfigService { private static final String secretKeyHeader = "fizz-secretkey"; - private Map app2gatewayGroupMap = new HashMap<>(32); + public Map serviceConfigMap = new HashMap<>(128); - private Map apiConfigMap = new HashMap<>(128); + private Map apiConfigMap = new HashMap<>(128); // TODO XXX @Value("${serviceWhiteList:x}") @@ -94,14 +94,17 @@ public class ApiConfigService { } } + @Value("${auth.compatible-wh:false}") + private boolean compatibleWh; + @Resource(name = AggregateRedisConfig.AGGREGATE_REACTIVE_REDIS_TEMPLATE) private ReactiveStringRedisTemplate rt; @Resource - private SystemConfig systemConfig; + private AppService appService; @Resource - private AppService appService; + private GatewayGroupService gatewayGroupService; @Autowired(required = false) private CustomAuth customAuth; @@ -127,7 +130,7 @@ public class ApiConfigService { try { ApiConfig ac = JacksonUtils.readValue(json, ApiConfig.class); apiConfigMap.put(ac.id, ac); - updateApp2gatewayGroupMap(ac); + updateServiceConfigMap(ac); return Flux.just(e); } catch (Throwable t) { throwable[0] = t; @@ -167,9 +170,9 @@ public class ApiConfigService { ApiConfig r = apiConfigMap.remove(ac.id); if (ac.isDeleted != ApiConfig.DELETED && r != null) { r.isDeleted = ApiConfig.DELETED; - updateApp2gatewayGroupMap(r); + updateServiceConfigMap(r); } - updateApp2gatewayGroupMap(ac); + updateServiceConfigMap(ac); if (ac.isDeleted != ApiConfig.DELETED) { apiConfigMap.put(ac.id, ac); } @@ -192,24 +195,21 @@ public class ApiConfigService { return Mono.just(ReactorUtils.EMPTY_THROWABLE); } - private void updateApp2gatewayGroupMap(ApiConfig ac) { - GatewayGroup gg = app2gatewayGroupMap.get(ac.app()); + private void updateServiceConfigMap(ApiConfig ac) { + ServiceConfig sc = serviceConfigMap.get(ac.service); if (ac.isDeleted == ApiConfig.DELETED) { - if (gg == null) { - log.info("no gateway group for " + ac.app()); + if (sc == null) { + log.info("no " + ac.service + " config to delete"); } else { - gg.remove(ac); - if (gg.getServiceConfigMap().isEmpty()) { - app2gatewayGroupMap.remove(ac.app()); - } + sc.remove(ac); } } else { - if (gg == null) { - gg = new GatewayGroup(ac.gatewayGroup); - app2gatewayGroupMap.put(ac.app(), gg); - gg.add(ac); + if (sc == null) { + sc = new ServiceConfig(ac.service); + serviceConfigMap.put(ac.service, sc); + sc.add(ac); } else { - gg.update(ac); + sc.update(ac); } } } @@ -218,13 +218,15 @@ public class ApiConfigService { YES (null), - CANT_ACCESS_CURRENT_GATEWAY_GROUP ("cant access current gateway group"), + NO_SERVICE_CONFIG ("no service config"), - NO_GATEWAY_GROUP_FOR_APP ("no gateway group for app"), + NO_API_CONFIG ("no api config"), - NO_APP_CONFIG_FOR_APP ("no app config for app"), + GATEWAY_GROUP_CANT_PROXY_API ("gateway group cant proxy api"), - ORIGIN_IP_NOT_IN_WHITE_LIST ("origin ip not in white list"), + APP_NOT_IN_API_LEGAL_APPS ("app not in api legal apps"), + + IP_NOT_IN_WHITE_LIST ("ip not in white list"), NO_TIMESTAMP_OR_SIGN ("no timestamp or sign"), @@ -236,10 +238,6 @@ public class ApiConfigService { SERVICE_NOT_OPEN ("service not open"), - NO_SERVICE_EXPOSE_TO_APP ("no service expose to app"), - - SERVICE_API_NOT_EXPOSE_TO_APP ("service api not expose to app"), - CANT_ACCESS_SERVICE_API ("cant access service api"); private String reason; @@ -254,7 +252,6 @@ public class ApiConfigService { } public Mono canAccess(ServerWebExchange exchange) { - ServerHttpRequest req = exchange.getRequest(); HttpHeaders hdrs = req.getHeaders(); LogService.setBizId(req.getId()); @@ -262,56 +259,83 @@ public class ApiConfigService { WebUtils.getServiceId(exchange), req.getMethod(), WebUtils.getReqPath(exchange)); } - private Mono canAccess(ServerWebExchange exchange, String app, String ip, String timestamp, String sign, String secretKey, - String service, HttpMethod method, String path) { + private Mono canAccess(ServerWebExchange exchange, String app, String ip, String timestamp, String sign, String secretKey, + String service, HttpMethod method, String path) { - GatewayGroup gg = app2gatewayGroupMap.get(app); boolean toCorBapp = App.TO_C.equals(app) || App.TO_B.equals(app); - - if (gg == null) { - if (toCorBapp) { return Mono.just(Access.YES); } else { return logWarnAndResult("no gateway group for " + app, Access.NO_GATEWAY_GROUP_FOR_APP); } + if (!whiteListSet.contains(service)) { // TODO XXX + return Mono.just(Access.SERVICE_NOT_OPEN); + } + ServiceConfig sc = serviceConfigMap.get(service); + if (sc == null) { + if (compatibleWh) { + return Mono.just(Access.YES); + } else { + return logWarnAndResult(service + Constants.Symbol.BLANK + Access.NO_SERVICE_CONFIG.getReason(), Access.NO_SERVICE_CONFIG); + } } else { - Set currentServerGatewayGroupSet = systemConfig.getCurrentServerGatewayGroupSet(); - if (currentServerGatewayGroupSet.contains(gg.id)) { - Mono am = Mono.just(Access.YES); - App a = appService.getApp(app); - if (a == null) { - if (!toCorBapp) { return logWarnAndResult("no app config for " + app, Access.NO_APP_CONFIG_FOR_APP); } - } else if (a.useWhiteList && !a.ips.contains(ip)) { - return logWarnAndResult(ip + " not in " + app + " white list", Access.ORIGIN_IP_NOT_IN_WHITE_LIST); - } else if (a.useAuth) { - if (a.authType == App.SIGN_AUTH) { - if (StringUtils.isBlank(timestamp) || StringUtils.isBlank(sign)) { return logWarnAndResult(app + " lack timestamp " + timestamp + " or sign " + sign, Access.NO_TIMESTAMP_OR_SIGN); } - else if (!validate(app, timestamp, a.secretkey, sign)) { return logWarnAndResult(app + " sign " + sign + " invalid", Access.SIGN_INVALID); } - } else if (customAuth == null) { - return logWarnAndResult(app + " no custom auth", Access.NO_CUSTOM_AUTH); - } else { - am = customAuth.auth(exchange, app, ip, timestamp, sign, secretKey, a); - } - } - return am.flatMap( - v -> { - LogService.setBizId(exchange.getRequest().getId()); - if (v == Access.CUSTOM_AUTH_REJECT || v != Access.YES) { return Mono.just(Access.CUSTOM_AUTH_REJECT); } - if (!whiteListSet.contains(service)) { return Mono.just(Access.SERVICE_NOT_OPEN); } // TODO XXX - ServiceConfig sc = gg.getServiceConfig(service); - if (sc == null) { - if (toCorBapp) { return Mono.just(Access.YES); } else { return logWarnAndResult("no service expose to " + app, Access.NO_SERVICE_EXPOSE_TO_APP); } - } else { - ApiConfig ac = sc.getApiConfig(method, path); - if (ac == null) { - if (toCorBapp) { return Mono.just(Access.YES); } else { return logWarnAndResult(service + ' ' + method.name() + ' ' + path + " not expose to " + app, Access.SERVICE_API_NOT_EXPOSE_TO_APP); } - } else if (ac.access == ApiConfig.ALLOW) { - return Mono.just(ac); - } else { - return logWarnAndResult(app + " cant access " + service + ' ' + method.name() + ' ' + path, Access.CANT_ACCESS_SERVICE_API); - } - } - } - ); - - } else { - return logWarnAndResult(app + " cant access " + currentServerGatewayGroupSet, Access.CANT_ACCESS_CURRENT_GATEWAY_GROUP); + String api = ThreadContext.getStringBuilder().append(service).append(Constants.Symbol.BLANK).append(method.name()).append(Constants.Symbol.BLANK + path).toString(); + ApiConfig ac0 = null; + for (String g : gatewayGroupService.currentGatewayGroupSet) { // compatible + ac0 = sc.getApiConfig(method, path, g, app); + if (ac0 != null) { + break; } + } + ApiConfig ac = ac0; + if (ac == null) { + if (compatibleWh) { + return Mono.just(Access.YES); + } else { + return logWarnAndResult(api + " no api config", Access.NO_API_CONFIG); + } + } else if (gatewayGroupService.currentGatewayGroupIn(ac.gatewayGroups)) { + if (ac.apps.contains(App.ALL_APP)) { + return allow(api, ac); + } else if (app != null && ac.apps.contains(app)) { + if (ac.access == ApiConfig.ALLOW) { + App a = appService.getApp(app); + if (a.useWhiteList && !a.allow(ip)) { + return logWarnAndResult(ip + " not in " + app + " white list", Access.IP_NOT_IN_WHITE_LIST); + } else if (a.useAuth) { + if (a.authType == App.SIGN_AUTH) { + if (StringUtils.isBlank(timestamp) || StringUtils.isBlank(sign)) { + return logWarnAndResult(app + " lack timestamp " + timestamp + " or sign " + sign, Access.NO_TIMESTAMP_OR_SIGN); + } else if (!validate(app, timestamp, a.secretkey, sign)) { + return logWarnAndResult(app + " sign " + sign + " invalid", Access.SIGN_INVALID); + } else { + return Mono.just(ac); + } + } else if (customAuth == null) { + return logWarnAndResult(app + " no custom auth", Access.NO_CUSTOM_AUTH); + } else { + return customAuth.auth(exchange, app, ip, timestamp, sign, secretKey, a).flatMap(v -> { + if (v == Access.YES) { + return Mono.just(ac); + } else { + return Mono.just(Access.CUSTOM_AUTH_REJECT); + } + }); + } + } else { + return Mono.just(ac); + } + } else { + return logWarnAndResult("cant access " + api, Access.CANT_ACCESS_SERVICE_API); + } + } else { + return logWarnAndResult(app + " not in " + api + " legal apps", Access.APP_NOT_IN_API_LEGAL_APPS); + } + } else { + return logWarnAndResult(gatewayGroupService.currentGatewayGroupSet + " cant proxy " + api, Access.GATEWAY_GROUP_CANT_PROXY_API); + } + } + } + + private static Mono allow(String api, ApiConfig ac) { + if (ac.access == ApiConfig.ALLOW) { + return Mono.just(ac); + } else { + return logWarnAndResult("cant access " + api, Access.CANT_ACCESS_SERVICE_API); } } @@ -321,12 +345,8 @@ public class ApiConfigService { } private static boolean validate(String app, String timestamp, String secretKey, String sign) { - StringBuilder b = new StringBuilder(128); + StringBuilder b = ThreadContext.getStringBuilder(); b.append(app).append(Constants.Symbol.UNDERLINE).append(timestamp).append(Constants.Symbol.UNDERLINE).append(secretKey); return sign.equals(DigestUtils.md532(b.toString())); } - - public Map getApp2gatewayGroupMap() { - return app2gatewayGroupMap; - } } diff --git a/src/main/java/we/plugin/auth/App.java b/src/main/java/we/plugin/auth/App.java index beaffc5..2e9e463 100644 --- a/src/main/java/we/plugin/auth/App.java +++ b/src/main/java/we/plugin/auth/App.java @@ -14,25 +14,24 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.plugin.auth; -import we.util.JacksonUtils; import org.apache.commons.lang3.StringUtils; -import java.util.Arrays; -import java.util.Collections; -import java.util.Set; +import we.util.Constants; +import we.util.JacksonUtils; + +import java.util.*; import java.util.stream.Collectors; /** - * @author lancer + * @author hongqiaowei */ public class App { - public static final String TO_C = "2c"; - - public static final String TO_B = "2b"; + public static final String ALL_APP = "*"; public static final int DELETED = 1; @@ -56,7 +55,9 @@ public class App { public boolean useWhiteList = false; - public Set ips = Collections.emptySet(); + public String config; + + private Map ips = new HashMap<>(6); public void setUseAuth(int i) { if (i == 1) { @@ -72,10 +73,69 @@ public class App { public void setIps(String ips) { if (StringUtils.isNotBlank(ips)) { - this.ips = Arrays.stream(StringUtils.split(ips, ',')).collect(Collectors.toSet()); + Arrays.stream(StringUtils.split(ips, ',')).forEach( + ip -> { + ip = ip.trim(); + int i = ip.lastIndexOf('.'); + String subnet = ip.substring(0, i).trim(); + String addrSeg = ip.substring(i + 1).trim(); + if ("*".equals(addrSeg)) { + this.ips.put(subnet, new String[]{"2", "254"}); + } else if (addrSeg.indexOf('-') > 0) { + String[] a = StringUtils.split(addrSeg, '-'); + String beg = a[0].trim(); + String end = a[1].trim(); + this.ips.put(subnet, new String[]{beg, end}); + } else { + this.ips.put(subnet, new String[]{addrSeg, addrSeg}); + } + } + ); } } + public boolean allow(String ip) { + int originSubnetLen = ip.lastIndexOf(Constants.Symbol.DOT); + for (Map.Entry e : ips.entrySet()) { + String subnet = e.getKey(); + int subnetLen = subnet.length(); + byte i = 0; + if (subnetLen == originSubnetLen) { + for (; i < subnetLen; i++) { + if (subnet.charAt(i) != ip.charAt(i)) { + break; + } + } + if (i == subnetLen) { + int originAddrLen = ip.length() - originSubnetLen - 1; + String[] addrSeg = e.getValue(); + String addrSegBeg = addrSeg[0]; + String addrSegEnd = addrSeg[1]; + if (originAddrLen < addrSegBeg.length() || addrSegEnd.length() < originAddrLen) { + return false; + } else { + if (originAddrLen == addrSegBeg.length()) { + for (byte j = 0; j < addrSegBeg.length(); j++) { + if (ip.charAt(originSubnetLen + 1 + j) < addrSegBeg.charAt(j)) { + return false; + } + } + } + if (originAddrLen == addrSegEnd.length()) { + for (byte j = 0; j < addrSegEnd.length(); j++) { + if (addrSegEnd.charAt(j) < ip.charAt(originSubnetLen + 1 + j)) { + return false; + } + } + } + return true; + } + } + } + } + return false; + } + @Override public String toString() { return JacksonUtils.writeValueAsString(this); diff --git a/src/main/java/we/plugin/auth/AppService.java b/src/main/java/we/plugin/auth/AppService.java index 06d4be0..7ce09e1 100644 --- a/src/main/java/we/plugin/auth/AppService.java +++ b/src/main/java/we/plugin/auth/AppService.java @@ -14,19 +14,20 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.plugin.auth; -import we.flume.clients.log4j2appender.LogService; -import we.listener.AggregateRedisConfig; -import we.util.Constants; -import we.util.JacksonUtils; -import we.util.ReactorUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.data.redis.core.ReactiveStringRedisTemplate; import org.springframework.stereotype.Service; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import we.flume.clients.log4j2appender.LogService; +import we.listener.AggregateRedisConfig; +import we.util.Constants; +import we.util.JacksonUtils; +import we.util.ReactorUtils; import javax.annotation.PostConstruct; import javax.annotation.Resource; @@ -37,7 +38,7 @@ import java.util.Objects; import java.util.concurrent.TimeUnit; /** - * @author lancer + * @author hongqiaowei */ @Service diff --git a/src/main/java/we/plugin/auth/AuthPluginFilter.java b/src/main/java/we/plugin/auth/AuthPluginFilter.java index 8dad4b3..43c5aa3 100644 --- a/src/main/java/we/plugin/auth/AuthPluginFilter.java +++ b/src/main/java/we/plugin/auth/AuthPluginFilter.java @@ -14,23 +14,24 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.plugin.auth; -import we.flume.clients.log4j2appender.LogService; -import we.plugin.PluginFilter; -import we.util.WebUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; import org.springframework.web.server.ServerWebExchange; import reactor.core.publisher.Mono; +import we.flume.clients.log4j2appender.LogService; +import we.plugin.PluginFilter; +import we.util.WebUtils; import javax.annotation.Resource; -import java.util.HashMap; +import java.util.Collections; import java.util.Map; /** - * @author lancer + * @author hongqiaowei */ @Component(AuthPluginFilter.AUTH_PLUGIN_FILTER) @@ -52,8 +53,7 @@ public class AuthPluginFilter extends PluginFilter { if (log.isDebugEnabled()) { log.debug("req auth: " + r, LogService.BIZ_ID, exchange.getRequest().getId()); } - Map data = new HashMap<>(1, 1.0f); - data.put(RESULT, r); + Map data = Collections.singletonMap(RESULT, r); return WebUtils.transmitSuccessFilterResultAndEmptyMono(exchange, AUTH_PLUGIN_FILTER, data); } ); diff --git a/src/main/java/we/plugin/auth/CustomAuth.java b/src/main/java/we/plugin/auth/CustomAuth.java index 9987724..59587a5 100644 --- a/src/main/java/we/plugin/auth/CustomAuth.java +++ b/src/main/java/we/plugin/auth/CustomAuth.java @@ -14,13 +14,14 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.plugin.auth; import org.springframework.web.server.ServerWebExchange; import reactor.core.publisher.Mono; /** - * @author lancer + * @author hongqiaowei */ public interface CustomAuth { diff --git a/src/main/java/we/plugin/auth/GatewayGroup.java b/src/main/java/we/plugin/auth/GatewayGroup.java index a632e75..2f58cbe 100644 --- a/src/main/java/we/plugin/auth/GatewayGroup.java +++ b/src/main/java/we/plugin/auth/GatewayGroup.java @@ -14,70 +14,52 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.plugin.auth; -import com.fasterxml.jackson.annotation.JsonIgnore; +import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import we.util.JacksonUtils; -import java.util.HashMap; -import java.util.Map; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; /** - * @author lancer + * @author hongqiaowei */ public class GatewayGroup { private static final Logger log = LoggerFactory.getLogger(GatewayGroup.class); - public static final char C = 'c'; + public static final String DEFAULT = "default"; - public static final char B = 'b'; + public static final int DELETED = 1; - public static final char T = 't'; + public int id; - public char id; + public int isDeleted = 0; - private Map serviceConfigMap = new HashMap<>(128); + public String group; - public GatewayGroup(char id) { - this.id = id; - } + public String name; - public Map getServiceConfigMap() { - return serviceConfigMap; - } + public Set gateways = new HashSet<>(); - @JsonIgnore - public ServiceConfig getServiceConfig(String id) { - return serviceConfigMap.get(id); - } - - public void remove(ApiConfig ac) { - ServiceConfig sc = serviceConfigMap.get(ac.service); - if (sc == null) { - log.info("no service config for " + ac); - } else { - sc.remove(ac); - if (sc.apiConfigMap().isEmpty()) { - serviceConfigMap.remove(ac.service); - } + public void setGateways(String gateways) { + if (StringUtils.isNotBlank(gateways)) { + Arrays.stream(StringUtils.split(gateways, ',')).forEach( + ip -> { + this.gateways.add(ip.trim()); + } + ); } } - public void add(ApiConfig ac) { - ServiceConfig sc = new ServiceConfig(ac.service); - serviceConfigMap.put(ac.service, sc); - sc.add(ac); - } - - public void update(ApiConfig ac) { - ServiceConfig sc = serviceConfigMap.get(ac.service); - if (sc == null) { - add(ac); - } else { - sc.update(ac); - } + @Override + public String toString() { + return JacksonUtils.writeValueAsString(this); } } diff --git a/src/main/java/we/plugin/auth/GatewayGroup2appsToApiConfig.java b/src/main/java/we/plugin/auth/GatewayGroup2appsToApiConfig.java new file mode 100644 index 0000000..91ff728 --- /dev/null +++ b/src/main/java/we/plugin/auth/GatewayGroup2appsToApiConfig.java @@ -0,0 +1,97 @@ +/* + * Copyright (C) 2020 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 . + */ + +package we.plugin.auth; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import we.util.JacksonUtils; + +import java.util.HashMap; +import java.util.Map; + +/** + * @author hongqiaowei + */ + +public class GatewayGroup2appsToApiConfig { + + private static final Logger log = LoggerFactory.getLogger(GatewayGroup2appsToApiConfig.class); + + private Map> configMap = new HashMap<>(6); + + public Map> getConfigMap() { + return configMap; + } + + public void setConfigMap(Map> configMap) { + this.configMap = configMap; + } + + public void add(ApiConfig ac) { + for (String gg : ac.gatewayGroups) { + Map app2apiConfigMap = configMap.get(gg); + if (app2apiConfigMap == null) { + app2apiConfigMap = new HashMap<>(); + configMap.put(gg, app2apiConfigMap); + } + for (String a : ac.apps) { + app2apiConfigMap.put(a, ac); + log.info(gg + " add " + a + " -> " + ac); + } + } + } + + public void remove(ApiConfig ac) { + for (String gg : ac.gatewayGroups) { + Map app2apiConfigMap = configMap.get(gg); + if (app2apiConfigMap != null) { + for (String a : ac.apps) { + ApiConfig r = app2apiConfigMap.remove(a); + log.info(gg + " remove " + a + " -> " + r); + } + } + } + } + + public void update(ApiConfig ac) { + for (String gg : ac.gatewayGroups) { + Map app2apiConfigMap = configMap.get(gg); + if (app2apiConfigMap == null) { + app2apiConfigMap = new HashMap<>(); + configMap.put(gg, app2apiConfigMap); + } + for (String a : ac.apps) { + ApiConfig old = app2apiConfigMap.put(a, ac); + log.info(gg + " update " + a + " -> " + old + " with " + ac); + } + } + } + + public ApiConfig get(String gatewayGroup, String app) { + Map app2apiConfigMap = configMap.get(gatewayGroup); + if (app2apiConfigMap == null) { + return null; + } else { + return app2apiConfigMap.get(app); + } + } + + public String toString() { + return JacksonUtils.writeValueAsString(this); + } +} diff --git a/src/main/java/we/plugin/auth/GatewayGroupService.java b/src/main/java/we/plugin/auth/GatewayGroupService.java new file mode 100644 index 0000000..a78c09d --- /dev/null +++ b/src/main/java/we/plugin/auth/GatewayGroupService.java @@ -0,0 +1,181 @@ +/* + * Copyright (C) 2020 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 . + */ + +package we.plugin.auth; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.data.redis.core.ReactiveStringRedisTemplate; +import org.springframework.stereotype.Service; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import we.flume.clients.log4j2appender.LogService; +import we.listener.AggregateRedisConfig; +import we.util.Constants; +import we.util.JacksonUtils; +import we.util.NetworkUtils; +import we.util.ReactorUtils; + +import javax.annotation.PostConstruct; +import javax.annotation.Resource; +import java.util.*; +import java.util.concurrent.TimeUnit; + +/** + * @author hongqiaowei + */ + +@Service +public class GatewayGroupService { + + private static final Logger log = LoggerFactory.getLogger(GatewayGroupService.class); + + private static final String fizzGatewayGroup = "fizz_gateway_group"; + + private static final String fizzGatewayGroupChannel = "fizz_gateway_group_channel"; + + public Map gatewayGroupMap = new HashMap<>(6); + + private Map oldGatewayGroupMap = new HashMap<>(6); + + public Set currentGatewayGroupSet = new HashSet<>(6); + + @Resource(name = AggregateRedisConfig.AGGREGATE_REACTIVE_REDIS_TEMPLATE) + private ReactiveStringRedisTemplate rt; + + @PostConstruct + public void init() throws Throwable { + final Throwable[] throwable = new Throwable[1]; + Throwable error = Mono.just(Objects.requireNonNull(rt.opsForHash().entries(fizzGatewayGroup) + .defaultIfEmpty(new AbstractMap.SimpleEntry<>(ReactorUtils.OBJ, ReactorUtils.OBJ)).onErrorStop().doOnError(t -> { + log.info(null, t); + }) + .concatMap(e -> { + Object k = e.getKey(); + if (k == ReactorUtils.OBJ) { + return Flux.just(e); + } + Object v = e.getValue(); + log.info(k.toString() + Constants.Symbol.COLON + v.toString(), LogService.BIZ_ID, k.toString()); + String json = (String) v; + try { + GatewayGroup gg = JacksonUtils.readValue(json, GatewayGroup.class); + oldGatewayGroupMap.put(gg.id, gg); + updateGatewayGroupMap(gg); + return Flux.just(e); + } catch (Throwable t) { + throwable[0] = t; + log.info(json, t); + return Flux.error(t); + } + }).blockLast())).flatMap( + e -> { + if (throwable[0] != null) { + return Mono.error(throwable[0]); + } + return lsnGatewayGroupChange(); + } + ).block(); + if (error != ReactorUtils.EMPTY_THROWABLE) { + throw error; + } + } + + private Mono lsnGatewayGroupChange() { + final Throwable[] throwable = new Throwable[1]; + final boolean[] b = {false}; + rt.listenToChannel(fizzGatewayGroupChannel).doOnError(t -> { + throwable[0] = t; + b[0] = false; + log.error("lsn " + fizzGatewayGroupChannel, t); + }).doOnSubscribe( + s -> { + b[0] = true; + log.info("success to lsn on " + fizzGatewayGroupChannel); + } + ).doOnNext(msg -> { + String json = msg.getMessage(); + log.info(json, LogService.BIZ_ID, "gg" + System.currentTimeMillis()); + try { + GatewayGroup gg = JacksonUtils.readValue(json, GatewayGroup.class); + GatewayGroup r = oldGatewayGroupMap.remove(gg.id); + if (gg.isDeleted != GatewayGroup.DELETED && r != null) { + gatewayGroupMap.remove(r.group); + } + updateGatewayGroupMap(gg); + if (gg.isDeleted != GatewayGroup.DELETED) { + oldGatewayGroupMap.put(gg.id, gg); + } + } catch (Throwable t) { + log.info(json, t); + } + }).subscribe(); + Throwable t = throwable[0]; + while (!b[0]) { + if (t != null) { + return Mono.error(t); + } else { + try { + TimeUnit.SECONDS.sleep(2); + } catch (InterruptedException e) { + return Mono.error(e); + } + } + } + return Mono.just(ReactorUtils.EMPTY_THROWABLE); + } + + private void updateGatewayGroupMap(GatewayGroup gg) { + if (gg.isDeleted == GatewayGroup.DELETED) { + GatewayGroup r = gatewayGroupMap.remove(gg.group); + log.info("remove " + r); + } else { + GatewayGroup existGatewayGroup = gatewayGroupMap.get(gg.group); + gatewayGroupMap.put(gg.group, gg); + if (existGatewayGroup == null) { + log.info("add " + gg); + } else { + log.info("update " + existGatewayGroup + " with " + gg); + } + } + updateCurrentGatewayGroupSet(); + } + + private void updateCurrentGatewayGroupSet() { + String ip = NetworkUtils.getServerIp(); + currentGatewayGroupSet.clear(); + gatewayGroupMap.forEach( + (k, gg) -> { + if (gg.gateways.contains(ip)) { + currentGatewayGroupSet.add(gg.group); + } + } + ); + if (currentGatewayGroupSet.isEmpty()) { + currentGatewayGroupSet.add(GatewayGroup.DEFAULT); + } + } + + public boolean currentGatewayGroupIn(Set gatewayGroups) { + for (String cgg : currentGatewayGroupSet) { + if (gatewayGroups.contains(cgg)) { + return true; + } + } + return false; + } +} diff --git a/src/main/java/we/plugin/auth/ServiceConfig.java b/src/main/java/we/plugin/auth/ServiceConfig.java index 6979b78..594a611 100644 --- a/src/main/java/we/plugin/auth/ServiceConfig.java +++ b/src/main/java/we/plugin/auth/ServiceConfig.java @@ -14,10 +14,14 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.plugin.auth; import com.fasterxml.jackson.annotation.JsonIgnore; + +import org.apache.commons.lang3.StringUtils; import we.util.Constants; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.http.HttpMethod; @@ -27,99 +31,122 @@ import java.util.HashMap; import java.util.Map; /** - * @author lancer + * @author hongqiaowei */ public class ServiceConfig { private static final Logger log = LoggerFactory.getLogger(ServiceConfig.class); - private static final String forward_slash_str = String.valueOf(Constants.Symbol.FORWARD_SLASH); + public String id; - public String id; + @JsonIgnore + public Map apiConfigMap = new HashMap<>(32); - private Map apiConfigMap = new HashMap<>(32); - - private Map> path2methodToApiConfigMapMap = new HashMap<>(32); + public Map> path2methodToApiConfigMapMap = new HashMap<>(6); public ServiceConfig(String id) { this.id = id; } - public Map apiConfigMap() { - return apiConfigMap; - } - - public Map> getPath2methodToApiConfigMapMap() { - return path2methodToApiConfigMapMap; + public void add(ApiConfig ac) { + apiConfigMap.put(ac.id, ac); + EnumMap method2apiConfigMap = path2methodToApiConfigMapMap.get(ac.path); + if (method2apiConfigMap == null) { + method2apiConfigMap = new EnumMap<>(HttpMethod.class); + GatewayGroup2appsToApiConfig gatewayGroup2appsToApiConfig = new GatewayGroup2appsToApiConfig(); + gatewayGroup2appsToApiConfig.add(ac); + method2apiConfigMap.put(ac.method, gatewayGroup2appsToApiConfig); + path2methodToApiConfigMapMap.put(ac.path, method2apiConfigMap); + } else { + GatewayGroup2appsToApiConfig gatewayGroup2appsToApiConfig = method2apiConfigMap.get(ac.method); + if (gatewayGroup2appsToApiConfig == null) { + gatewayGroup2appsToApiConfig = new GatewayGroup2appsToApiConfig(); + method2apiConfigMap.put(ac.method, gatewayGroup2appsToApiConfig); + } + gatewayGroup2appsToApiConfig.add(ac); + } + log.info("add " + ac); } public void remove(ApiConfig ac) { ApiConfig remove = apiConfigMap.remove(ac.id); - log.info(remove + " is removed from api config map"); - Map method2apiConfigMap = path2methodToApiConfigMapMap.get(ac.path); - if (method2apiConfigMap != null) { - ApiConfig r = method2apiConfigMap.remove(ac.method); - log.info(r + " is removed from method 2 api config map"); - if (method2apiConfigMap.isEmpty()) { - path2methodToApiConfigMapMap.remove(ac.path); - } - } else { - log.info("no method 2 api config map for " + ac.path); - } - } - - public void add(ApiConfig ac) { - apiConfigMap.put(ac.id, ac); - EnumMap method2apiConfigMap = path2methodToApiConfigMapMap.get(ac.path); + Map method2apiConfigMap = path2methodToApiConfigMapMap.get(ac.path); if (method2apiConfigMap == null) { - method2apiConfigMap = new EnumMap<>(HttpMethod.class); - path2methodToApiConfigMapMap.put(ac.path, method2apiConfigMap); + log.info("no config to delete for " + ac.service + ' ' + ac.path); + } else { + GatewayGroup2appsToApiConfig gatewayGroup2appsToApiConfig = method2apiConfigMap.get(ac.method); + if (gatewayGroup2appsToApiConfig == null) { + log.info("no config to delete for " + ac.service + ' ' + ac.method + ' ' + ac.path); + } else { + log.info(id + " remove " + ac); + gatewayGroup2appsToApiConfig.remove(ac); + } } - method2apiConfigMap.put(ac.method, ac); - log.info(ac + " is added to api config map"); } public void update(ApiConfig ac) { ApiConfig prev = apiConfigMap.put(ac.id, ac); log.info(prev + " is updated by " + ac + " in api config map"); - EnumMap method2apiConfigMap = path2methodToApiConfigMapMap.get(ac.path); + EnumMap method2apiConfigMap = path2methodToApiConfigMapMap.get(ac.path); if (method2apiConfigMap == null) { method2apiConfigMap = new EnumMap<>(HttpMethod.class); + GatewayGroup2appsToApiConfig gatewayGroup2appsToApiConfig = new GatewayGroup2appsToApiConfig(); + gatewayGroup2appsToApiConfig.add(ac); + method2apiConfigMap.put(ac.method, gatewayGroup2appsToApiConfig); path2methodToApiConfigMapMap.put(ac.path, method2apiConfigMap); + } else { + GatewayGroup2appsToApiConfig gatewayGroup2appsToApiConfig = method2apiConfigMap.get(ac.method); + if (gatewayGroup2appsToApiConfig == null) { + gatewayGroup2appsToApiConfig = new GatewayGroup2appsToApiConfig(); + method2apiConfigMap.put(ac.method, gatewayGroup2appsToApiConfig); + gatewayGroup2appsToApiConfig.add(ac); + } else { + log.info(id + " update " + ac); + gatewayGroup2appsToApiConfig.update(ac); + } } - ApiConfig put = method2apiConfigMap.put(ac.method, ac); - log.info(put + " is updated by " + ac + " in method 2 api config map"); } @JsonIgnore - public ApiConfig getApiConfig(HttpMethod method, String path) { + public ApiConfig getApiConfig(HttpMethod method, String path, String gatewayGroup, String app) { + GatewayGroup2appsToApiConfig r = getApiConfig0(method, path); + if (r == null) { + return null; + } + if (StringUtils.isBlank(app)) { + app = App.ALL_APP; + } + return r.get(gatewayGroup, app); + } + + private GatewayGroup2appsToApiConfig getApiConfig0(HttpMethod method, String path) { while (true) { - EnumMap method2apiConfigMap = path2methodToApiConfigMapMap.get(path); + EnumMap method2apiConfigMap = path2methodToApiConfigMapMap.get(path); if (method2apiConfigMap == null) { int i = path.lastIndexOf(Constants.Symbol.FORWARD_SLASH); if (i == 0) { - method2apiConfigMap = path2methodToApiConfigMapMap.get(forward_slash_str); + method2apiConfigMap = path2methodToApiConfigMapMap.get(Constants.Symbol.FORWARD_SLASH_STR); if (method2apiConfigMap == null) { return null; } else { - return getApiConfig0(method, method2apiConfigMap); + return getApiConfig1(method, method2apiConfigMap); } } else { path = path.substring(0, i); } } else { - return getApiConfig0(method, method2apiConfigMap); + return getApiConfig1(method, method2apiConfigMap); } } } - private ApiConfig getApiConfig0(HttpMethod method, EnumMap method2apiConfigMap) { - ApiConfig ac = method2apiConfigMap.get(method); - if (ac == null) { + private GatewayGroup2appsToApiConfig getApiConfig1(HttpMethod method, EnumMap method2apiConfigMap) { + GatewayGroup2appsToApiConfig r = method2apiConfigMap.get(method); + if (r == null) { return method2apiConfigMap.get(HttpMethod.X); } else { - return ac; + return r; } } } diff --git a/src/main/java/we/plugin/stat/StatPluginFilter.java b/src/main/java/we/plugin/stat/StatPluginFilter.java index 24bca24..e90557a 100644 --- a/src/main/java/we/plugin/stat/StatPluginFilter.java +++ b/src/main/java/we/plugin/stat/StatPluginFilter.java @@ -14,24 +14,32 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.plugin.stat; -import we.flume.clients.log4j2appender.LogService; -import we.plugin.PluginFilter; -import we.util.Constants; -import we.util.ThreadContext; -import we.util.WebUtils; +import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Value; +import org.springframework.data.redis.core.ReactiveStringRedisTemplate; import org.springframework.stereotype.Component; import org.springframework.web.server.ServerWebExchange; import reactor.core.publisher.Mono; +import we.flume.clients.log4j2appender.LogService; +import we.listener.AggregateRedisConfig; +import we.plugin.PluginFilter; +import we.plugin.auth.GatewayGroupService; +import we.util.Constants; +import we.util.ThreadContext; +import we.util.WebUtils; +import javax.annotation.PostConstruct; +import javax.annotation.Resource; +import java.util.Iterator; import java.util.Map; /** - * @author lancer + * @author hongqiaowei */ @Component(StatPluginFilter.STAT_PLUGIN_FILTER) @@ -41,8 +49,6 @@ public class StatPluginFilter extends PluginFilter { public static final String STAT_PLUGIN_FILTER = "statPlugin"; - private static final String accessStat = "$accessStat"; - private static final String ip = "\"ip\":"; private static final String gatewayGroup = "\"gatewayGroup\":"; @@ -60,24 +66,37 @@ public class StatPluginFilter extends PluginFilter { @Value("${stat.open:false}") private boolean statOpen = false; - @Value("${stat.topic:fizz_access_stat}") + @Value("${stat.channel:fizz_access_stat}") + private String fizzAccessStatChannel; + + @Value("${stat.topic:}") private String fizzAccessStatTopic; + @Resource(name = AggregateRedisConfig.AGGREGATE_REACTIVE_REDIS_TEMPLATE) + private ReactiveStringRedisTemplate rt; + + @Resource + private GatewayGroupService gatewayGroupService; + + private String currentGatewayGroups; + + @PostConstruct + public void init() { + Iterator it = gatewayGroupService.currentGatewayGroupSet.iterator(); + currentGatewayGroups = it.next(); + while (it.hasNext()) { + currentGatewayGroups = currentGatewayGroups + ',' + it.next(); + } + } + @Override public Mono doFilter(ServerWebExchange exchange, Map config, String fixedConfig) { if (statOpen) { - StringBuilder b = (StringBuilder) ThreadContext.get(accessStat); - if (b == null) { - b = new StringBuilder(128); - ThreadContext.set(accessStat, b); - } else { - b.delete(0, b.length()); - } - + StringBuilder b = ThreadContext.getStringBuilder(); b.append(Constants.Symbol.LEFT_BRACE); b.append(ip); toJsonStringValue(b, WebUtils.getOriginIp(exchange)); b.append(Constants.Symbol.COMMA); - b.append(gatewayGroup); toJsonStringValue(b, WebUtils.getCurrentGatewayGroup(exchange)); b.append(Constants.Symbol.COMMA); + b.append(gatewayGroup); toJsonStringValue(b, currentGatewayGroups); b.append(Constants.Symbol.COMMA); b.append(service); toJsonStringValue(b, WebUtils.getServiceId(exchange)); b.append(Constants.Symbol.COMMA); b.append(appid); toJsonStringValue(b, WebUtils.getAppId(exchange)); b.append(Constants.Symbol.COMMA); b.append(apiMethod); toJsonStringValue(b, exchange.getRequest().getMethodValue()); b.append(Constants.Symbol.COMMA); @@ -85,7 +104,11 @@ public class StatPluginFilter extends PluginFilter { b.append(reqTime) .append(System.currentTimeMillis()); b.append(Constants.Symbol.RIGHT_BRACE); - log.info(b.toString(), LogService.HANDLE_STGY, LogService.toKF(fizzAccessStatTopic)); + if (StringUtils.isBlank(fizzAccessStatTopic)) { + rt.convertAndSend(fizzAccessStatChannel, b.toString()); + } else { + log.info(b.toString(), LogService.HANDLE_STGY, LogService.toKF(fizzAccessStatTopic)); + } } return WebUtils.transmitSuccessFilterResultAndEmptyMono(exchange, STAT_PLUGIN_FILTER, null); diff --git a/src/main/java/we/proxy/CallBackendConfig.java b/src/main/java/we/proxy/CallBackendConfig.java index 7d61112..f7e60b2 100644 --- a/src/main/java/we/proxy/CallBackendConfig.java +++ b/src/main/java/we/proxy/CallBackendConfig.java @@ -14,10 +14,11 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.proxy; /** - * @author lancer + * @author hongqiaowei * 请求转发/调用后端接口时的负载均衡、流控、failover、超时等配置 */ diff --git a/src/main/java/we/proxy/FizzWebClient.java b/src/main/java/we/proxy/FizzWebClient.java index 348385e..d1be262 100644 --- a/src/main/java/we/proxy/FizzWebClient.java +++ b/src/main/java/we/proxy/FizzWebClient.java @@ -14,17 +14,13 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.proxy; import com.netflix.appinfo.InstanceInfo; import com.netflix.discovery.EurekaClient; import com.netflix.discovery.shared.Applications; -import we.flume.clients.log4j2appender.LogService; -import we.config.AggrWebClientConfig; -import we.config.ProxyWebClientConfig; -import we.util.Constants; -import we.util.ThreadContext; -import we.util.WebUtils; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Value; @@ -38,14 +34,22 @@ import org.springframework.web.reactive.function.client.ClientResponse; import org.springframework.web.reactive.function.client.WebClient; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import we.config.AggrWebClientConfig; +import we.config.ProxyWebClientConfig; +import we.flume.clients.log4j2appender.LogService; +import we.util.Constants; +import we.util.ThreadContext; +import we.util.WebUtils; import javax.annotation.PostConstruct; import javax.annotation.Resource; +import java.util.ArrayList; import java.util.List; import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.atomic.AtomicLong; /** - * @author lancer + * @author hongqiaowei */ @Service @@ -95,18 +99,26 @@ public class FizzWebClient { return aggrResolveAddressSend(aggrService, aggrMethod, aggrPath, originReqIdOrBizId, method, uriOrSvc, headers, body, null); } - private Mono aggrResolveAddressSend(String aggrService, HttpMethod aggrMethod, String aggrPath, @Nullable String originReqIdOrBizId, - HttpMethod method, String uriOrSvc, @Nullable HttpHeaders headers, @Nullable Object body, @Nullable CallBackendConfig cbc) { + public Mono send(String reqId, HttpMethod method, String uriOrSvc, @Nullable HttpHeaders headers, @Nullable Object body) { + return send(reqId, method, uriOrSvc, headers, body, null); + } + public Mono send(String reqId, HttpMethod method, String uriOrSvc, HttpHeaders headers, Object body, CallBackendConfig cbc) { String s = extractServiceOrAddress(uriOrSvc); if (isService(s)) { String path = uriOrSvc.substring(uriOrSvc.indexOf(Constants.Symbol.FORWARD_SLASH, 10)); - return send2service(originReqIdOrBizId, method, s, path, headers, body, cbc); + return send2service(reqId, method, s, path, headers, body, cbc); } else { - return send2uri(originReqIdOrBizId, method, uriOrSvc, headers, body, cbc); + return send2uri(reqId, method, uriOrSvc, headers, body, cbc); } } + private Mono aggrResolveAddressSend(String aggrService, HttpMethod aggrMethod, String aggrPath, @Nullable String originReqIdOrBizId, + HttpMethod method, String uriOrSvc, @Nullable HttpHeaders headers, @Nullable Object body, @Nullable CallBackendConfig cbc) { + + return send(originReqIdOrBizId, method, uriOrSvc, headers, body, cbc); + } + public Mono proxySend2service(@Nullable String originReqIdOrBizId, HttpMethod method, String service, String relativeUri, @Nullable HttpHeaders headers, @Nullable Object body) { @@ -218,7 +230,23 @@ public class FizzWebClient { return b.append(Constants.Symbol.HTTP_PROTOCOL_PREFIX).append(inst.getIPAddr()).append(Constants.Symbol.COLON).append(inst.getPort()).append(path).toString(); } + + // private static List aggrMemberInsts = new ArrayList<>(); + // static { + // InstanceInfo i0 = InstanceInfo.Builder.newBuilder().setAppName("TRIP-MINI").setIPAddr("xxx.25.63.192").setPort(7094).build(); + // aggrMemberInsts.add(i0); + // } + // private static AtomicLong counter = new AtomicLong(0); + // private static final String aggrMember = "trip-mini"; + + private InstanceInfo roundRobinChoose1instFrom(String service) { + + // if (aggrMember.equals(service)) { + // int idx = (int) (counter.incrementAndGet() % aggrMemberInsts.size()); + // return aggrMemberInsts.get(idx); + // } + List insts = eurekaClient.getInstancesByVipAddress(service, false); if (insts == null || insts.isEmpty()) { throw new RuntimeException("eureka no " + service, null, false, false) {}; diff --git a/src/main/java/we/util/Constants.java b/src/main/java/we/util/Constants.java index 3152b26..b17404f 100644 --- a/src/main/java/we/util/Constants.java +++ b/src/main/java/we/util/Constants.java @@ -14,12 +14,13 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.util; import org.apache.commons.lang3.SystemUtils; /** - * @author lancer + * @author hongqiaowei */ public final class Constants { @@ -33,6 +34,7 @@ public final class Constants { public static final char BLANK = ' '; public static final char SPACE = BLANK; public static final char FORWARD_SLASH = '/'; + public static final String FORWARD_SLASH_STR = "/"; public static final char BACK_SLASH = '\\'; public static final char DOT = '.'; public static final char SEMICOLON = ';'; diff --git a/src/main/java/we/util/DateTimeUtils.java b/src/main/java/we/util/DateTimeUtils.java index 2b2a20d..511e5cc 100644 --- a/src/main/java/we/util/DateTimeUtils.java +++ b/src/main/java/we/util/DateTimeUtils.java @@ -14,9 +14,8 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ -package we.util; -import we.util.Constants.DatetimePattern; +package we.util; import java.time.*; import java.time.format.DateTimeFormatter; @@ -25,8 +24,10 @@ import java.util.*; import java.util.stream.Collectors; import java.util.stream.Stream; +import we.util.Constants.DatetimePattern; + /** - * @author lancer + * @author hongqiaowei */ public abstract class DateTimeUtils { diff --git a/src/main/java/we/util/DigestUtils.java b/src/main/java/we/util/DigestUtils.java index 0f89972..ca4466d 100644 --- a/src/main/java/we/util/DigestUtils.java +++ b/src/main/java/we/util/DigestUtils.java @@ -22,7 +22,7 @@ import org.apache.commons.codec.binary.Hex; import java.security.MessageDigest; /** - * @author lancer + * @author hongqiaowei */ public abstract class DigestUtils extends org.apache.commons.codec.digest.DigestUtils { diff --git a/src/main/java/we/util/JacksonUtils.java b/src/main/java/we/util/JacksonUtils.java index c4dce6d..0b186bf 100644 --- a/src/main/java/we/util/JacksonUtils.java +++ b/src/main/java/we/util/JacksonUtils.java @@ -14,6 +14,7 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.util; import com.fasterxml.jackson.annotation.JsonInclude.Include; @@ -23,6 +24,7 @@ import com.fasterxml.jackson.core.JsonParser; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.*; import com.fasterxml.jackson.databind.module.SimpleModule; + import we.plugin.auth.ApiConfig; import we.util.Constants.DatetimePattern; @@ -33,7 +35,7 @@ import java.time.format.DateTimeFormatter; import java.util.Date; /** - * @author lancer + * @author hongqiaowei */ public abstract class JacksonUtils { @@ -42,19 +44,19 @@ public abstract class JacksonUtils { static { JsonFactory f = new JsonFactory(); - f.configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true); + f.configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true); f.configure(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES, true); m = new ObjectMapper(f); m.setSerializationInclusion(Include.NON_EMPTY); - m.configure(SerializationFeature.WRITE_ENUMS_USING_TO_STRING, true); - m.configure(DeserializationFeature.READ_ENUMS_USING_TO_STRING, true); - m.configure(DeserializationFeature.FAIL_ON_NUMBERS_FOR_ENUMS, true); - m.configure(SerializationFeature.WRITE_EMPTY_JSON_ARRAYS, true); // FIXME - m.configure(SerializationFeature.WRITE_NULL_MAP_VALUES, true); - m.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); - m.configure(JsonParser.Feature.ALLOW_UNQUOTED_CONTROL_CHARS, true); + m.configure( SerializationFeature. WRITE_ENUMS_USING_TO_STRING, true); + m.configure( DeserializationFeature. READ_ENUMS_USING_TO_STRING, true); + m.configure( DeserializationFeature. FAIL_ON_NUMBERS_FOR_ENUMS, true); + m.configure( SerializationFeature. WRITE_EMPTY_JSON_ARRAYS, true); // FIXME + m.configure( SerializationFeature. WRITE_NULL_MAP_VALUES, true); + m.configure( DeserializationFeature. FAIL_ON_UNKNOWN_PROPERTIES, false); + m.configure( JsonParser.Feature. ALLOW_UNQUOTED_CONTROL_CHARS, true); SimpleModule m0 = new SimpleModule(); m0.addDeserializer(Date.class, new DateDeseralizer()); diff --git a/src/main/java/we/util/NetworkUtils.java b/src/main/java/we/util/NetworkUtils.java index 5d8d503..579a35c 100644 --- a/src/main/java/we/util/NetworkUtils.java +++ b/src/main/java/we/util/NetworkUtils.java @@ -14,6 +14,7 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.util; import org.slf4j.Logger; @@ -27,7 +28,7 @@ import java.security.SecureRandom; import java.util.Enumeration; /** - * @author lancer + * @author hongqiaowei */ public class NetworkUtils { @@ -75,7 +76,7 @@ public class NetworkUtils { public static int getServerId() { if (serverId == -1) { try { - StringBuilder b = new StringBuilder(); + StringBuilder b = ThreadContext.getStringBuilder(); Enumeration nis = NetworkInterface.getNetworkInterfaces(); while (nis.hasMoreElements()) { NetworkInterface ni = nis.nextElement(); diff --git a/src/main/java/we/util/ReactorUtils.java b/src/main/java/we/util/ReactorUtils.java index 58f5700..d8f73ee 100644 --- a/src/main/java/we/util/ReactorUtils.java +++ b/src/main/java/we/util/ReactorUtils.java @@ -14,12 +14,13 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.util; import reactor.core.publisher.Mono; /** - * @author lancer + * @author hongqiaowei */ public interface ReactorUtils { diff --git a/src/main/java/we/util/Script.java b/src/main/java/we/util/Script.java index ceede2f..fd1e791 100644 --- a/src/main/java/we/util/Script.java +++ b/src/main/java/we/util/Script.java @@ -14,10 +14,11 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.util; /** - * @author lancer + * @author hongqiaowei */ public class Script { diff --git a/src/main/java/we/util/ScriptUtils.java b/src/main/java/we/util/ScriptUtils.java index b00afd7..ba0c76a 100644 --- a/src/main/java/we/util/ScriptUtils.java +++ b/src/main/java/we/util/ScriptUtils.java @@ -14,9 +14,9 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.util; -import com.fasterxml.jackson.core.JsonProcessingException; import org.apache.commons.lang3.StringUtils; import javax.script.*; @@ -26,7 +26,7 @@ import java.util.HashMap; import java.util.Map; /** - * @author lancer + * @author hongqiaowei */ public abstract class ScriptUtils { diff --git a/src/main/java/we/util/ThreadContext.java b/src/main/java/we/util/ThreadContext.java index ffb53ec..f654f47 100644 --- a/src/main/java/we/util/ThreadContext.java +++ b/src/main/java/we/util/ThreadContext.java @@ -14,6 +14,7 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.util; import java.text.SimpleDateFormat; @@ -21,7 +22,7 @@ import java.util.HashMap; import java.util.Map; /** - * @author lancer + * @author hongqiaowei */ public abstract class ThreadContext { @@ -29,7 +30,7 @@ public abstract class ThreadContext { private static ThreadLocal> tl = new ThreadLocal<>(); private static final int mapCap = 32; - private static final String sb = "_sb"; + private static final String sb = "$sb"; private static final int sbCap = 256; /** use me carefully! */ @@ -39,31 +40,29 @@ public abstract class ThreadContext { /** use me carefully! */ public static StringBuilder getStringBuilder(boolean clean) { - // Map m = getMap(); - // StringBuilder b = (StringBuilder) m.get(sb); - // if (b == null) { - // b = new StringBuilder(sbCap); - // m.put(sb, b); - // } else { - // if (clean) { - // b.delete(0, b.length()); - // } - // } - // return b; - return new StringBuilder(64); + Map m = getMap(); + StringBuilder b = (StringBuilder) m.get(sb); + if (b == null) { + b = new StringBuilder(sbCap); + m.put(sb, b); + } else { + if (clean) { + b.delete(0, b.length()); + } + } + return b; } public static StringBuilder getStringBuilder(String key) { - // StringBuilder b = (StringBuilder) get(key); - // if (b == null) { - // b = new StringBuilder(sbCap); - // Map m = getMap(); - // m.put(key, b); - // } else { - // b.delete(0, b.length()); - // } - // return b; - return getStringBuilder(true); + StringBuilder b = (StringBuilder) get(key); + if (b == null) { + b = new StringBuilder(sbCap); + Map m = getMap(); + m.put(key, b); + } else { + b.delete(0, b.length()); + } + return b; } /** for legacy code. */ diff --git a/src/main/java/we/util/Utils.java b/src/main/java/we/util/Utils.java index 71d7438..709d643 100644 --- a/src/main/java/we/util/Utils.java +++ b/src/main/java/we/util/Utils.java @@ -14,6 +14,7 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.util; import org.apache.commons.lang3.StringUtils; @@ -23,7 +24,7 @@ import java.time.LocalDateTime; import java.time.LocalTime; /** - * @author lancer + * @author hongqiaowei */ public abstract class Utils { diff --git a/src/main/java/we/util/WebUtils.java b/src/main/java/we/util/WebUtils.java index f09b2f7..2e5525d 100644 --- a/src/main/java/we/util/WebUtils.java +++ b/src/main/java/we/util/WebUtils.java @@ -14,11 +14,9 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ + package we.util; -import we.flume.clients.log4j2appender.LogService; -import we.filter.FilterResult; -import we.legacy.RespEntity; import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -32,11 +30,14 @@ import org.springframework.lang.Nullable; import org.springframework.web.reactive.function.client.ClientResponse; import org.springframework.web.server.ServerWebExchange; import reactor.core.publisher.Mono; +import we.filter.FilterResult; +import we.flume.clients.log4j2appender.LogService; +import we.legacy.RespEntity; import java.util.*; /** - * @author lancer + * @author hongqiaowei */ public abstract class WebUtils { @@ -57,7 +58,7 @@ public abstract class WebUtils { private static final String SERVICE_ID = "serviceId"; - private static final String xForwardedFor = "X_FORWARDED_FOR"; + private static final String xForwardedFor = "X-FORWARDED-FOR"; private static final String unknown = "unknown"; @@ -73,8 +74,6 @@ public abstract class WebUtils { private static final String originIp = "originIp"; - public static final String CGG = "cgg"; - public static String getHeaderValue(ServerWebExchange exchange, String header) { return exchange.getRequest().getHeaders().getFirst(header); } @@ -87,10 +86,6 @@ public abstract class WebUtils { return exchange.getAttribute(APP_HEADER); } - public static Character getCurrentGatewayGroup(ServerWebExchange exchange) { - return exchange.getAttribute(CGG); - } - public static String getServiceId(ServerWebExchange exchange) { String svc = exchange.getAttribute(SERVICE_ID); if (svc == null) {