add sentinel cluster flow client demo (#6)

src/adservice-v2/src/main/java/org/daocloud/springcloud/adservice/AdserviceApplication.java

@@ -1,12 +1,15 @@
 package org.daocloud.springcloud.adservice;
 
+import org.daocloud.springcloud.adservice.init.SentinelClusterClientInitFunc;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
 import org.springframework.cloud.client.loadbalancer.LoadBalanced;
 import org.springframework.context.annotation.Bean;
 import org.springframework.web.client.RestTemplate;
 
+import javax.annotation.PostConstruct;
+
 @SpringBootApplication
 @EnableDiscoveryClient
 public class AdserviceApplication {
@@ -19,5 +22,10 @@ public RestTemplate restTemplate() {
  public static void main(String[] args) {
  SpringApplication.run(AdserviceApplication.class, args);
  }
+
+ @PostConstruct
+ public void initSentinelClusterFlow() throws Exception{
+ new SentinelClusterClientInitFunc().init();
+ }
 }
 

src/adservice-v2/src/main/java/org/daocloud/springcloud/adservice/consts/SentinelConst.java

@@ -0,0 +1,8 @@
+package org.daocloud.springcloud.adservice.consts;
+
+public final class SentinelConst {
+ public static final String FLOW_POSTFIX = "-flow-rules";
+ public static final String PARAM_FLOW_POSTFIX = "-param-rules";
+ public static final String CLUSTER_CLIENT_POSTFIX = "-cluster-client-config";
+ public static final String CLUSTER_MAP_POSTFIX = "-cluster-map";
+}

src/adservice-v2/src/main/java/org/daocloud/springcloud/adservice/controller/SentinelRuleDemo.java

@@ -113,4 +113,9 @@ private void initAuthorityRule() {
  rules.add(rule);
  AuthorityRuleManager.loadRules(rules);
  }
+
+ @GetMapping("/cluster-flow")
+ public String clusterFlow() {
+ return sentinelRuleDemoService.clusterFlow();
+ }
 }

src/adservice-v2/src/main/java/org/daocloud/springcloud/adservice/dto/ClusterGroupDto.java

@@ -0,0 +1,52 @@
+package org.daocloud.springcloud.adservice.dto;
+
+import java.util.Set;
+
+public class ClusterGroupDto {
+ private String machineId;
+ private String ip;
+ private Integer port;
+ private Set<String> clientSet;
+
+ public String getMachineId() {
+ return machineId;
+ }
+
+ public void setMachineId(String machineId) {
+ this.machineId = machineId;
+ }
+
+ public String getIp() {
+ return ip;
+ }
+
+ public void setIp(String ip) {
+ this.ip = ip;
+ }
+
+ public Integer getPort() {
+ return port;
+ }
+
+ public void setPort(Integer port) {
+ this.port = port;
+ }
+
+ public Set<String> getClientSet() {
+ return clientSet;
+ }
+
+ public void setClientSet(Set<String> clientSet) {
+ this.clientSet = clientSet;
+ }
+
+ @Override
+ public String toString() {
+ return "ClusterGroupDto{" +
+ "machineId='" + machineId + '\'' +
+ ", ip='" + ip + '\'' +
+ ", port=" + port +
+ ", clientSet=" + clientSet +
+ '}';
+ }
+}

src/adservice-v2/src/main/java/org/daocloud/springcloud/adservice/init/SentinelClusterClientInitFunc.java

@@ -0,0 +1,138 @@
+package org.daocloud.springcloud.adservice.init;
+
+import com.alibaba.csp.sentinel.cluster.ClusterStateManager;
+import com.alibaba.csp.sentinel.cluster.client.config.ClusterClientAssignConfig;
+import com.alibaba.csp.sentinel.cluster.client.config.ClusterClientConfig;
+import com.alibaba.csp.sentinel.cluster.client.config.ClusterClientConfigManager;
+import com.alibaba.csp.sentinel.datasource.ReadableDataSource;
+import com.alibaba.csp.sentinel.datasource.nacos.NacosDataSource;
+import com.alibaba.csp.sentinel.init.InitFunc;
+import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
+import com.alibaba.csp.sentinel.slots.block.flow.FlowRuleManager;
+import com.alibaba.csp.sentinel.slots.block.flow.param.ParamFlowRule;
+import com.alibaba.csp.sentinel.slots.block.flow.param.ParamFlowRuleManager;
+import com.alibaba.csp.sentinel.transport.config.TransportConfig;
+import com.alibaba.csp.sentinel.util.AppNameUtil;
+import com.alibaba.csp.sentinel.util.HostNameUtil;
+import com.alibaba.csp.sentinel.util.StringUtil;
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.TypeReference;
+import org.daocloud.springcloud.adservice.consts.SentinelConst;
+import org.daocloud.springcloud.adservice.dto.ClusterGroupDto;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+public class SentinelClusterClientInitFunc implements InitFunc {
+
+ private static final String APP_NAME = AppNameUtil.getAppName();
+
+ private String nacosAddress;
+ private final String groupId = "SENTINEL_GROUP";
+
+ private final String flowDataId = APP_NAME + SentinelConst.FLOW_POSTFIX;
+ private final String paramDataId = APP_NAME + SentinelConst.PARAM_FLOW_POSTFIX;
+ private final String clusterClientConfig = APP_NAME + SentinelConst.CLUSTER_CLIENT_POSTFIX;
+ private final String clusterMapDataId = APP_NAME + SentinelConst.CLUSTER_MAP_POSTFIX;
+
+ @Override
+ public void init() throws Exception {
+ nacosAddress = System.getProperty("spring.cloud.nacos.config.server-addr");
+ if (StringUtil.isBlank(nacosAddress)){
+ throw new RuntimeException("nacos address start param must be set");
+ }
+ System.out.printf("nacos address: %s\n", nacosAddress);
+
+ initDynamicRuleProperty();
+
+ initClientConfigProperty();
+
+ initClientServerAssignProperty();
+
+ initStateProperty();
+ }
+
+ private void initDynamicRuleProperty() {
+ ReadableDataSource<String, List<FlowRule>> ruleSource = new NacosDataSource<>(nacosAddress, groupId,
+ flowDataId, source -> JSON.parseObject(source, new TypeReference<List<FlowRule>>() {}));
+ FlowRuleManager.register2Property(ruleSource.getProperty());
+
+ ReadableDataSource<String, List<ParamFlowRule>> paramRuleSource = new NacosDataSource<>(nacosAddress, groupId,
+ paramDataId, source -> JSON.parseObject(source, new TypeReference<List<ParamFlowRule>>() {}));
+ ParamFlowRuleManager.register2Property(paramRuleSource.getProperty());
+ }
+
+ private void initClientConfigProperty() {
+ ReadableDataSource<String, ClusterClientConfig> clientConfigDs = new NacosDataSource<>(nacosAddress, groupId,
+ clusterClientConfig, source -> JSON.parseObject(source, new TypeReference<ClusterClientConfig>() {}));
+ ClusterClientConfigManager.registerClientConfigProperty(clientConfigDs.getProperty());
+ }
+
+ private void initClientServerAssignProperty() {
+// Cluster map format:
+// [
+// {
+// "machineId": "10.64.0.81@8720",
+// "ip": "10.64.0.81",
+// "port": 18730,
+// "clientSet": ["10.64.0.81@8721", "10.64.0.81@8722"]
+// }
+// ]
+ ReadableDataSource<String, ClusterClientAssignConfig> clientAssignDs = new NacosDataSource<>(nacosAddress, groupId,
+ clusterMapDataId, source -> {
+ List<ClusterGroupDto> groupList = JSON.parseObject(source, new TypeReference<List<ClusterGroupDto>>() {});
+ return Optional.ofNullable(groupList)
+ .flatMap(this::extractClientAssignment)
+ .orElse(null);
+ });
+ ClusterClientConfigManager.registerServerAssignProperty(clientAssignDs.getProperty());
+ }
+
+ private void initStateProperty() {
+ ReadableDataSource<String, Integer> clusterModeDs = new NacosDataSource<>(nacosAddress, groupId,
+ clusterMapDataId, source -> {
+ List<ClusterGroupDto> groupList = JSON.parseObject(source, new TypeReference<List<ClusterGroupDto>>() {});
+ return Optional.ofNullable(groupList)
+ .map(this::extractMode)
+ .orElse(ClusterStateManager.CLUSTER_NOT_STARTED);
+ });
+ ClusterStateManager.registerProperty(clusterModeDs.getProperty());
+ }
+
+ private int extractMode(List<ClusterGroupDto> groupList) {
+ if (groupList.stream().anyMatch(this::machineEqual)) {
+ return ClusterStateManager.CLUSTER_SERVER;
+ }
+
+ boolean canBeClient = groupList.stream()
+ .flatMap(e -> e.getClientSet().stream())
+ .filter(Objects::nonNull)
+ .anyMatch(e -> e.equals(getCurrentMachineId()));
+ return canBeClient ? ClusterStateManager.CLUSTER_CLIENT : ClusterStateManager.CLUSTER_NOT_STARTED;
+ }
+
+ private Optional<ClusterClientAssignConfig> extractClientAssignment(List<ClusterGroupDto> groupList) {
+ if (groupList.stream().anyMatch(this::machineEqual)) {
+ return Optional.empty();
+ }
+ for (ClusterGroupDto group : groupList) {
+ if (group.getClientSet().contains(getCurrentMachineId())) {
+ String ip = group.getIp();
+ Integer port = group.getPort();
+ return Optional.of(new ClusterClientAssignConfig(ip, port));
+ }
+ }
+ return Optional.empty();
+ }
+
+ private boolean machineEqual(ClusterGroupDto group) {
+ return getCurrentMachineId().equals(group.getMachineId());
+ }
+
+ private String getCurrentMachineId() {
+ return HostNameUtil.getIp() + SEPARATOR + TransportConfig.getRuntimePort();
+ }
+
+ private static final String SEPARATOR = "@";
+}

src/adservice-v2/src/main/java/org/daocloud/springcloud/adservice/service/SentinelRuleDemoService.java

@@ -1,6 +1,7 @@
 package org.daocloud.springcloud.adservice.service;
 
 import com.alibaba.csp.sentinel.annotation.SentinelResource;
+import com.alibaba.csp.sentinel.slots.block.BlockException;
 import org.springframework.stereotype.Service;
 
 @Service
@@ -9,4 +10,14 @@ public class SentinelRuleDemoService {
  public String common() {
  return "common";
  }
+
+ @SentinelResource(value = "cluster-resource", blockHandler = "clusterFlowBlockHandler")
+ public String clusterFlow() {
+ return "cluster flow test";
+ }
+
+ public String clusterFlowBlockHandler(BlockException ex) {
+ ex.printStackTrace();
+ return String.format("Oops, <%s> blocked by Sentinel", "cluster flow test");
+ }
 }