Add Apollo data source extension (#46)

pom.xml

@@ -101,6 +101,11 @@
                 <artifactId>sentinel-datasource-zookeeper</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>com.alibaba.csp</groupId>
+                <artifactId>sentinel-datasource-apollo</artifactId>
+                <version>${project.version}</version>
+            </dependency>
             <dependency>
                 <groupId>com.alibaba.csp</groupId>
                 <artifactId>sentinel-transport-simple-http</artifactId>

sentinel-demo/pom.xml

@@ -19,6 +19,7 @@
         <module>sentinel-demo-dubbo</module>
         <module>sentinel-demo-nacos-datasource</module>
         <module>sentinel-demo-zookeeper-datasource</module>
+        <module>sentinel-demo-apollo-datasource</module>
         <module>sentinel-demo-annotation-spring-aop</module>
     </modules>
 

sentinel-demo/sentinel-demo-apollo-datasource/pom.xml

@@ -0,0 +1,63 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <parent>
+    <artifactId>sentinel-demo</artifactId>
+    <groupId>com.alibaba.csp</groupId>
+    <version>0.1.1-SNAPSHOT</version>
+  </parent>
+  <modelVersion>4.0.0</modelVersion>
+
+  <artifactId>sentinel-demo-apollo-datasource</artifactId>
+
+  <properties>
+    <java.source.version>1.8</java.source.version>
+    <java.target.version>1.8</java.target.version>
+    <log4j2.version>2.9.1</log4j2.version>
+    <slf4j.version>1.7.25</slf4j.version>
+  </properties>
+
+  <dependencyManagement>
+    <dependencies>
+      <dependency>
+        <groupId>org.slf4j</groupId>
+        <artifactId>slf4j-api</artifactId>
+        <version>${slf4j.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>org.apache.logging.log4j</groupId>
+        <artifactId>log4j-core</artifactId>
+        <version>${log4j2.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>org.apache.logging.log4j</groupId>
+        <artifactId>log4j-api</artifactId>
+        <version>${log4j2.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>org.apache.logging.log4j</groupId>
+        <artifactId>log4j-slf4j-impl</artifactId>
+        <version>${log4j2.version}</version>
+      </dependency>
+    </dependencies>
+  </dependencyManagement>
+
+  <dependencies>
+    <dependency>
+      <groupId>com.alibaba.csp</groupId>
+      <artifactId>sentinel-datasource-apollo</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>com.alibaba</groupId>
+      <artifactId>fastjson</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.logging.log4j</groupId>
+      <artifactId>log4j-slf4j-impl</artifactId>
+    </dependency>
+  </dependencies>
+
+</project>

sentinel-demo/sentinel-demo-apollo-datasource/src/main/java/com/alibaba/csp/sentinel/demo/datasource/apollo/ApolloDataSourceDemo.java

@@ -0,0 +1,70 @@
+package com.alibaba.csp.sentinel.demo.datasource.apollo;
+
+import com.alibaba.csp.sentinel.datasource.DataSource;
+import com.alibaba.csp.sentinel.datasource.apollo.ApolloDataSource;
+import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
+import com.alibaba.csp.sentinel.slots.block.flow.FlowRuleManager;
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.TypeReference;
+import java.util.List;
+
+/**
+ * This demo shows how to use Apollo as the data source of Sentinel rules.
+ * <br />
+ * You need to first set up data as follows:
+ * <ol>
+ *  <li>Create an application with app id as sentinel-demo in Apollo</li>
+ *  <li>
+ *    Create a configuration with key as flowRules and value as follows:
+ *    <pre>
+ *      [
+          {
+            "resource": "TestResource",
+            "controlBehavior": 0,
+            "count": 5.0,
+            "grade": 1,
+            "limitApp": "default",
+            "strategy": 0
+          }
+        ]
+ *    </pre>
+ *  </li>
+ *  <li>Publish the application namespace</li>
+ * </ol>
+ * Then you could start this demo and adjust the rule configuration as you wish.
+ * The rule changes will take effect in real time.
+ *
+ * @author Jason Song
+ */
+public class ApolloDataSourceDemo {
+  private static final String KEY = "TestResource";
+
+  public static void main(String[] args) {
+    loadRules();
+    // Assume we config: resource is `TestResource`, initial QPS threshold is 5.
+    FlowQpsRunner runner = new FlowQpsRunner(KEY, 1, 100);
+    runner.simulateTraffic();
+    runner.tick();
+  }
+
+  private static void loadRules() {
+    /**
+     * Set up basic information, only for demo purpose. You may adjust them based on your actual environment.
+     * <br />
+     * For more information, please refer https://github.com/ctripcorp/apollo
+     */
+    String appId = "sentinel-demo";
+    String apolloMetaServerAddress = "http://localhost:8080";
+    System.setProperty("app.id", appId);
+    System.setProperty("apollo.meta", apolloMetaServerAddress);
+
+    String namespaceName = "application";
+    String flowRuleKey = "flowRules";
+    String defaultFlowRules = "[]"; //it's better to provide a meaningful default value
+
+    DataSource<String, List<FlowRule>> flowRuleDataSource = new ApolloDataSource<>(namespaceName, flowRuleKey,
+        defaultFlowRules, source -> JSON.parseObject(source, new TypeReference<List<FlowRule>>() {
+    }));
+    FlowRuleManager.register2Property(flowRuleDataSource.getProperty());
+  }
+}

sentinel-demo/sentinel-demo-apollo-datasource/src/main/java/com/alibaba/csp/sentinel/demo/datasource/apollo/FlowQpsRunner.java

@@ -0,0 +1,139 @@
+/*
+ * Copyright 1999-2018 Alibaba Group Holding Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.alibaba.csp.sentinel.demo.datasource.apollo;
+
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.alibaba.csp.sentinel.Entry;
+import com.alibaba.csp.sentinel.SphU;
+import com.alibaba.csp.sentinel.slots.block.BlockException;
+import com.alibaba.csp.sentinel.util.TimeUtil;
+
+/**
+ * Flow QPS runner.
+ *
+ * @author Carpenter Lee
+ * @author Eric Zhao
+ */
+class FlowQpsRunner {
+
+    private final String resourceName;
+    private final int threadCount;
+    private int seconds;
+
+    public FlowQpsRunner(String resourceName, int threadCount, int seconds) {
+        this.resourceName = resourceName;
+        this.threadCount = threadCount;
+        this.seconds = seconds;
+    }
+
+    private final AtomicInteger pass = new AtomicInteger();
+    private final AtomicInteger block = new AtomicInteger();
+    private final AtomicInteger total = new AtomicInteger();
+
+    private volatile boolean stop = false;
+
+    public void simulateTraffic() {
+        for (int i = 0; i < threadCount; i++) {
+            Thread t = new Thread(new RunTask());
+            t.setName("simulate-traffic-Task");
+            t.start();
+        }
+    }
+
+    public void tick() {
+        Thread timer = new Thread(new TimerTask());
+        timer.setName("sentinel-timer-task");
+        timer.start();
+    }
+
+    final class RunTask implements Runnable {
+        @Override
+        public void run() {
+            while (!stop) {
+                Entry entry = null;
+
+                try {
+                    entry = SphU.entry(resourceName);
+                    // token acquired, means pass
+                    pass.addAndGet(1);
+                } catch (BlockException e1) {
+                    block.incrementAndGet();
+                } catch (Exception e2) {
+                    // biz exception
+                } finally {
+                    total.incrementAndGet();
+                    if (entry != null) {
+                        entry.exit();
+                    }
+                }
+
+                Random random2 = new Random();
+                try {
+                    TimeUnit.MILLISECONDS.sleep(random2.nextInt(50));
+                } catch (InterruptedException e) {
+                    // ignore
+                }
+            }
+        }
+    }
+
+    final class TimerTask implements Runnable {
+        @Override
+        public void run() {
+            long start = System.currentTimeMillis();
+            System.out.println("begin to statistic!!!");
+
+            long oldTotal = 0;
+            long oldPass = 0;
+            long oldBlock = 0;
+            while (!stop) {
+                try {
+                    TimeUnit.SECONDS.sleep(1);
+                } catch (InterruptedException e) {
+                }
+                long globalTotal = total.get();
+                long oneSecondTotal = globalTotal - oldTotal;
+                oldTotal = globalTotal;
+
+                long globalPass = pass.get();
+                long oneSecondPass = globalPass - oldPass;
+                oldPass = globalPass;
+
+                long globalBlock = block.get();
+                long oneSecondBlock = globalBlock - oldBlock;
+                oldBlock = globalBlock;
+
+                System.out.println(seconds + " send qps is: " + oneSecondTotal);
+                System.out.println(TimeUtil.currentTimeMillis() + ", total:" + oneSecondTotal
+                    + ", pass:" + oneSecondPass
+                    + ", block:" + oneSecondBlock);
+
+                if (seconds-- <= 0) {
+                    stop = true;
+                }
+            }
+
+            long cost = System.currentTimeMillis() - start;
+            System.out.println("time cost: " + cost + " ms");
+            System.out.println("total:" + total.get() + ", pass:" + pass.get()
+                + ", block:" + block.get());
+            System.exit(0);
+        }
+    }
+}

sentinel-demo/sentinel-demo-apollo-datasource/src/main/resources/log4j2.xml

@@ -0,0 +1,16 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration monitorInterval="60">
+  <appenders>
+    <Console name="Console" target="SYSTEM_OUT">
+      <PatternLayout pattern="[apollo-datasource-demo][%t]%d %-5p [%c] %m%n"/>
+    </Console>
+    <Async name="Async" includeLocation="true">
+      <AppenderRef ref="Console"/>
+    </Async>
+  </appenders>
+  <loggers>
+    <root level="INFO">
+      <AppenderRef ref="Async"/>
+    </root>
+  </loggers>
+</configuration>

sentinel-extension/pom.xml

@@ -15,7 +15,7 @@
         <module>sentinel-datasource-extension</module>
         <module>sentinel-datasource-nacos</module>
         <module>sentinel-datasource-zookeeper</module>
-
+        <module>sentinel-datasource-apollo</module>
         <module>sentinel-annotation-aspectj</module>
     </modules>
 

sentinel-extension/sentinel-datasource-apollo/pom.xml

@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <parent>
+    <artifactId>sentinel-extension</artifactId>
+    <groupId>com.alibaba.csp</groupId>
+    <version>0.1.1-SNAPSHOT</version>
+  </parent>
+  <modelVersion>4.0.0</modelVersion>
+
+  <artifactId>sentinel-datasource-apollo</artifactId>
+
+  <properties>
+    <apollo.version>1.0.0</apollo.version>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>com.alibaba.csp</groupId>
+      <artifactId>sentinel-datasource-extension</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>com.ctrip.framework.apollo</groupId>
+      <artifactId>apollo-client</artifactId>
+      <version>${apollo.version}</version>
+    </dependency>
+  </dependencies>
+
+</project>

sentinel-extension/sentinel-datasource-apollo/src/main/java/com/alibaba/csp/sentinel/datasource/apollo/ApolloDataSource.java

@@ -0,0 +1,94 @@
+package com.alibaba.csp.sentinel.datasource.apollo;
+
+import com.alibaba.csp.sentinel.datasource.AbstractDataSource;
+import com.alibaba.csp.sentinel.datasource.ConfigParser;
+import com.alibaba.csp.sentinel.datasource.DataSource;
+import com.alibaba.csp.sentinel.log.RecordLog;
+import com.ctrip.framework.apollo.Config;
+import com.ctrip.framework.apollo.ConfigChangeListener;
+import com.ctrip.framework.apollo.ConfigService;
+import com.ctrip.framework.apollo.model.ConfigChange;
+import com.ctrip.framework.apollo.model.ConfigChangeEvent;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+import com.google.common.collect.Sets;
+
+/**
+ * A {@link DataSource} with <a href="http://github.com/ctripcorp/apollo">Apollo</a> as its configuration source.
+ * <br />
+ * When the rule is changed in Apollo, it will take effect in real time.
+ *
+ * @author Jason Song
+ */
+public class ApolloDataSource<T> extends AbstractDataSource<String, T> {
+
+  private final Config config;
+  private final String flowRulesKey;
+  private final String defaultFlowRuleValue;
+
+  /**
+   * Constructs the Apollo data source
+   *
+   * @param namespaceName        the namespace name in Apollo, should not be null or empty
+   * @param flowRulesKey         the flow rules key in the namespace, should not be null or empty
+   * @param defaultFlowRuleValue the default flow rules value when the flow rules key is not found or any error occurred
+   * @param parser               the parser to transform string configuration to actual flow rules
+   */
+  public ApolloDataSource(String namespaceName, String flowRulesKey, String defaultFlowRuleValue,
+      ConfigParser<String, T> parser) {
+    super(parser);
+
+    Preconditions.checkArgument(!Strings.isNullOrEmpty(namespaceName), "Namespace name could not be null or empty");
+    Preconditions.checkArgument(!Strings.isNullOrEmpty(flowRulesKey), "FlowRuleKey could not be null or empty!");
+
+    this.flowRulesKey = flowRulesKey;
+    this.defaultFlowRuleValue = defaultFlowRuleValue;
+
+    this.config = ConfigService.getConfig(namespaceName);
+
+    initialize();
+
+    RecordLog.info(String.format("Initialized rule for namespace: %s, flow rules key: %s", namespaceName, flowRulesKey));
+  }
+
+  private void initialize() {
+    initializeConfigChangeListener();
+    loadAndUpdateRules();
+  }
+
+  private void loadAndUpdateRules() {
+    try {
+      T newValue = loadConfig();
+      if (newValue == null) {
+        RecordLog.warn("[ApolloDataSource] WARN: rule config is null, you may have to check your data source");
+      }
+      getProperty().updateValue(newValue);
+    } catch (Throwable ex) {
+      RecordLog.warn("[ApolloDataSource] Error when loading rule config", ex);
+    }
+  }
+
+  private void initializeConfigChangeListener() {
+    config.addChangeListener(new ConfigChangeListener() {
+      @Override
+      public void onChange(ConfigChangeEvent changeEvent) {
+        ConfigChange change = changeEvent.getChange(flowRulesKey);
+        //change is never null because the listener will only notify for this key
+        if (change != null) {
+          RecordLog.info("[ApolloDataSource] Received config changes: " + change.toString());
+        }
+        loadAndUpdateRules();
+      }
+    }, Sets.newHashSet(flowRulesKey));
+  }
+
+  @Override
+  public String readSource() throws Exception {
+    return config.getProperty(flowRulesKey, defaultFlowRuleValue);
+  }
+
+  @Override
+  public void close() throws Exception {
+    // nothing to destroy
+  }
+}