diff --git a/fizz-bootstrap/src/main/resources/application.yml b/fizz-bootstrap/src/main/resources/application.yml index 71bc049..d29cc87 100644 --- a/fizz-bootstrap/src/main/resources/application.yml +++ b/fizz-bootstrap/src/main/resources/application.yml @@ -103,6 +103,7 @@ refresh-local-cache: app-auth-enabled: true flow-control-rule-enabled: true rpc-service-enabled: true + degrade-rule-enabled: true fizz: diff --git a/fizz-core/src/main/java/we/config/RefreshLocalCacheConfig.java b/fizz-core/src/main/java/we/config/RefreshLocalCacheConfig.java index 0c3559d..496a708 100644 --- a/fizz-core/src/main/java/we/config/RefreshLocalCacheConfig.java +++ b/fizz-core/src/main/java/we/config/RefreshLocalCacheConfig.java @@ -26,6 +26,7 @@ import we.plugin.auth.ApiConfig2appsService; import we.plugin.auth.AppService; import we.plugin.auth.GatewayGroupService; import we.proxy.RpcInstanceService; +import we.stats.degrade.DegradeRuleService; import we.stats.ratelimit.ResourceRateLimitConfigService; import javax.annotation.Resource; @@ -39,6 +40,7 @@ import javax.annotation.Resource; * @see AppService#refreshLocalCache() refresh app local cache * @see ResourceRateLimitConfigService#refreshLocalCache() refresh flow control rule local cache * @see RpcInstanceService#refreshLocalCache() refresh rpc service local cache + * @see DegradeRuleService#refreshLocalCache() refresh degrade rule local cache * * @author zhongjie */ @@ -73,6 +75,9 @@ public class RefreshLocalCacheConfig { @Resource private FizzMangerConfig fizzMangerConfig; + @Resource + private DegradeRuleService degradeRuleService; + @Scheduled(initialDelayString = "${refresh-local-cache.initial-delay-millis:300000}", fixedRateString = "${refresh-local-cache.fixed-rate-millis:300000}") public void refreshLocalCache() { @@ -139,6 +144,15 @@ public class RefreshLocalCacheConfig { } } + if (refreshLocalCacheConfigProperties.isDegradeRuleCacheRefreshEnabled()) { + LOGGER.debug("refresh degrade rule local cache"); + try { + degradeRuleService.refreshLocalCache(); + } catch (Throwable t) { + LOGGER.warn("refresh degrade rule local cache exception", t); + } + } + fizzMangerConfig.updateMangerUrl(); } } diff --git a/fizz-core/src/main/java/we/config/RefreshLocalCacheConfigProperties.java b/fizz-core/src/main/java/we/config/RefreshLocalCacheConfigProperties.java index dd17bac..fdff10c 100644 --- a/fizz-core/src/main/java/we/config/RefreshLocalCacheConfigProperties.java +++ b/fizz-core/src/main/java/we/config/RefreshLocalCacheConfigProperties.java @@ -51,4 +51,7 @@ public class RefreshLocalCacheConfigProperties { @Value("${refresh-local-cache.rpc-service-enabled:false}") private boolean rpcServiceCacheRefreshEnabled; + + @Value("${refresh-local-cache.degrade-rule-enabled:false}") + private boolean degradeRuleCacheRefreshEnabled; } diff --git a/fizz-core/src/main/java/we/stats/degrade/DegradeRule.java b/fizz-core/src/main/java/we/stats/degrade/DegradeRule.java new file mode 100644 index 0000000..b7ef705 --- /dev/null +++ b/fizz-core/src/main/java/we/stats/degrade/DegradeRule.java @@ -0,0 +1,141 @@ +/* + * Copyright (C) 2021 the original author or authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ +package we.stats.degrade; + +import com.fasterxml.jackson.annotation.JsonIgnore; +import lombok.Data; +import org.apache.commons.lang3.StringUtils; +import we.util.JacksonUtils; + +import static we.util.ResourceIdUtils.SERVICE_DEFAULT; + +/** + * Degrade rule entity + * + * @author zhongjie + */ +@Data +public class DegradeRule { + /** + * 整型自增主键 + */ + private Long id; + /** + * 熔断类型 1-服务默认配置 2-服务 3-接口 + */ + private Byte type; + /** + * 前端服务名 + */ + private String service; + /** + * 前端API路径 + */ + private String path; + /** + * 熔断策略 1-异常比例 2-异常数 + */ + private Byte strategy; + /** + * 比例阈值，当熔断策略为 1-异常比例 时该字段有值 + */ + private Float ratioThreshold; + /** + * 异常数，当熔断策略为 2-异常数 时该字段有值 + */ + private Long exceptionCount; + /** + * 最小请求数 + */ + private Long minRequestCount; + /** + * 熔断时长（秒） + */ + private Integer timeWindow; + /** + * 统计时长（秒） + */ + private Integer statInterval; + /** + * 恢复策略 1-尝试恢复 2-逐步恢复 3-立即恢复 + */ + private Byte recoveryStrategy; + /** + * 恢复时长（秒），当恢复策略为 2-逐步恢复 时该字段有值 + */ + private Integer recoveryTimeWindow; + /** + * 熔断响应ContentType + */ + private String responseContentType; + /** + * 熔断响应报文 + */ + private String responseContent; + + /** + * 当type为1时，1启用，0反之 + */ + private Integer enable; + /** + * 是否删除 1-是 2-否 + */ + private Integer isDeleted; + + private String resourceId; + + public boolean isDeleted() { + return isDeleted == 1; + } + + public boolean isEnable() { + return enable == 1; + } + + public void setType(Byte type) { + this.type = type; + if (type == 1) { + service = SERVICE_DEFAULT; + } + } + + public void setService(String s) { + if (StringUtils.isNotBlank(s)) { + service = s; + } + } + + public void setPath(String p) { + if (StringUtils.isNotBlank(p)) { + path = p; + } + } + + + @JsonIgnore + public String getResourceId() { + if (resourceId == null) { + resourceId = "^^^" + (service == null ? "" : service) + '^' + (path == null ? "" : path); + } + return resourceId; + } + + @Override + public String toString() { + return JacksonUtils.writeValueAsString(this); + } +} diff --git a/fizz-core/src/main/java/we/stats/degrade/DegradeRuleService.java b/fizz-core/src/main/java/we/stats/degrade/DegradeRuleService.java new file mode 100644 index 0000000..3870080 --- /dev/null +++ b/fizz-core/src/main/java/we/stats/degrade/DegradeRuleService.java @@ -0,0 +1,164 @@ +/* + * Copyright (C) 2021 the original author or authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ +package we.stats.degrade; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.data.redis.core.ReactiveStringRedisTemplate; +import org.springframework.stereotype.Service; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import we.config.AggregateRedisConfig; +import we.util.JacksonUtils; +import we.util.Result; + +import javax.annotation.PostConstruct; +import javax.annotation.Resource; +import java.util.Collections; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Degrade rule service + * + * @author zhongjie + */ +@Service +@Slf4j +public class DegradeRuleService { + + /** + * Redis degrade rule change channel + */ + private static final String DEGRADE_RULE_CHANNEL = "fizz_degrade_rule_channel"; + /** + * redis degrade rule info hash key + */ + private static final String DEGRADE_RULE_HASH_KEY = "fizz_degrade_rule"; + + + @Resource(name = AggregateRedisConfig.AGGREGATE_REACTIVE_REDIS_TEMPLATE) + private ReactiveStringRedisTemplate rt; + + private Map resourceId2DegradeRuleMap = new ConcurrentHashMap<>(32); + private Map id2DegradeRuleMap = new ConcurrentHashMap<>(32); + + @PostConstruct + public void init() { + Result result = initDegradeRule(); + if (result.code == Result.FAIL) { + throw new RuntimeException(result.msg, result.t); + } + result = lsnDegradeRuleChange(); + if (result.code == Result.FAIL) { + throw new RuntimeException(result.msg, result.t); + } + } + + public void refreshLocalCache() throws Throwable { + this.initDegradeRule(); + } + + private Result initDegradeRule() { + Result result = Result.succ(); + + Map resourceId2DegradeRuleMapTmp = new ConcurrentHashMap<>(32); + Map id2DegradeRuleMapTmp = new ConcurrentHashMap<>(32); + + Flux> degradeRuleEntries = rt.opsForHash().entries(DEGRADE_RULE_HASH_KEY); + degradeRuleEntries.collectList() + .defaultIfEmpty(Collections.emptyList()) + .flatMap(es -> { + if (!es.isEmpty()) { + String json = null; + try { + for (Map.Entry e : es) { + json = (String) e.getValue(); + DegradeRule degradeRule = JacksonUtils.readValue(json, DegradeRule.class); + resourceId2DegradeRuleMapTmp.put(degradeRule.getResourceId(), degradeRule); + id2DegradeRuleMapTmp.put(degradeRule.getId(), degradeRule); + log.info("init degrade rule: {}", json); + } + } catch (Throwable t) { + result.code = Result.FAIL; + result.msg = "init degrade rule error, json: " + json; + result.t = t; + } + } else { + log.info("no degrade rule"); + } + return Mono.empty(); + } + ) + .onErrorReturn( + throwable -> { + result.code = Result.FAIL; + result.msg = "init degrade rule error"; + result.t = throwable; + return true; + }, + result + ) + .block(); + resourceId2DegradeRuleMap = resourceId2DegradeRuleMapTmp; + id2DegradeRuleMap = id2DegradeRuleMapTmp; + return result; + } + + private Result lsnDegradeRuleChange() { + Result result = Result.succ(); + rt.listenToChannel(DEGRADE_RULE_CHANNEL) + .doOnError( + t -> { + result.code = Result.FAIL; + result.msg = "lsn error, channel: " + DEGRADE_RULE_CHANNEL; + result.t = t; + log.error("lsn channel {} error", DEGRADE_RULE_CHANNEL, t); + } + ) + .doOnSubscribe( + s -> log.info("success to lsn on {}", DEGRADE_RULE_CHANNEL) + ) + .doOnNext( + msg -> { + String message = msg.getMessage(); + try { + DegradeRule degradeRule = JacksonUtils.readValue(message, DegradeRule.class); + if (degradeRule.isDeleted()) { + DegradeRule remove = id2DegradeRuleMap.remove(degradeRule.getId()); + if (remove != null) { + resourceId2DegradeRuleMap.remove(remove.getResourceId()); + } + log.info("remove degrade rule {}", message); + } else { + DegradeRule previous = id2DegradeRuleMap.put(degradeRule.getId(), degradeRule); + if (previous != null) { + if (!previous.getResourceId().equals(degradeRule.getResourceId())) { + resourceId2DegradeRuleMap.remove(previous.getResourceId()); + } + } + resourceId2DegradeRuleMap.put(degradeRule.getResourceId(), degradeRule); + log.info("update degrade rule {}", message); + } + } catch (Throwable t) { + log.error("update degrade rule error, {}", message, t); + } + } + ) + .subscribe(); + return result; + } +}