Add Etcd data-source extension (#1018)

pom.xml

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

sentinel-demo/pom.xml

@@ -34,6 +34,7 @@
         <module>sentinel-demo-apache-dubbo</module>
         <module>sentinel-demo-spring-cloud-gateway</module>
         <module>sentinel-demo-zuul-gateway</module>
+        <module>sentinel-demo-etcd-datasource</module>
     </modules>
 
     <dependencies>

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

@@ -0,0 +1,48 @@
+<?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>1.7.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>sentinel-demo-etcd-datasource</artifactId>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>com.alibaba.csp</groupId>
+            <artifactId>sentinel-core</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.alibaba.csp</groupId>
+            <artifactId>sentinel-datasource-etcd</artifactId>
+        </dependency>
+
+
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+        </dependency>
+
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <version>${maven.compiler.version}</version>
+                <configuration>
+                    <source>1.8</source>
+                    <target>1.8</target>
+                    <encoding>${java.encoding}</encoding>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+</project>

sentinel-demo/sentinel-demo-etcd-datasource/src/main/java/com/alibaba/csp/sentinel/demo/datasource/etcd/EtcdConfigSender.java

@@ -0,0 +1,58 @@
+/*
+ * 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.etcd;
+
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+
+/**
+ * Etcd config sender for demo.
+ *
+ * @author lianglin
+ * @since 1.7.0
+ */
+public class EtcdConfigSender {
+
+    public static void main(String[] args) throws InterruptedException {
+
+
+        String rule_key = "sentinel_demo_rule_key";
+
+        Client client = Client.builder()
+                .endpoints("http://127.0.0.1:2379")
+                .user(ByteSequence.from("root".getBytes()))
+                .password(ByteSequence.from("12345".getBytes()))
+                .build();
+        final String rule = "[\n"
+                + "  {\n"
+                + "    \"resource\": \"TestResource\",\n"
+                + "    \"controlBehavior\": 0,\n"
+                + "    \"count\": 5.0,\n"
+                + "    \"grade\": 1,\n"
+                + "    \"limitApp\": \"default\",\n"
+                + "    \"strategy\": 0\n"
+                + "  }\n"
+                + "]";
+        client.getKVClient()
+                .put(ByteSequence.from(rule_key.getBytes()), ByteSequence.from(rule.getBytes()));
+
+        System.out.println("setting rule success");
+        Thread.sleep(10000);
+
+    }
+
+}

sentinel-demo/sentinel-demo-etcd-datasource/src/main/java/com/alibaba/csp/sentinel/demo/datasource/etcd/EtcdDataSourceDemo.java

@@ -0,0 +1,52 @@
+/*
+ * 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.etcd;
+
+import com.alibaba.csp.sentinel.config.SentinelConfig;
+import com.alibaba.csp.sentinel.datasource.ReadableDataSource;
+import com.alibaba.csp.sentinel.datasource.etcd.EtcdConfig;
+import com.alibaba.csp.sentinel.datasource.etcd.EtcdDataSource;
+import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
+import com.alibaba.csp.sentinel.slots.block.flow.FlowRuleManager;
+import com.alibaba.fastjson.JSON;
+
+import java.util.List;
+
+/**
+ * @author lianglin
+ * @since 1.7.0
+ */
+public class EtcdDataSourceDemo {
+
+    public static void main(String[] args) {
+
+        String rule_key = "sentinel_demo_rule_key";
+        String yourUserName = "root";
+        String yourPassWord = "12345";
+        String endPoints = "http://127.0.0.1:2379";
+        SentinelConfig.setConfig(EtcdConfig.END_POINTS, endPoints);
+        SentinelConfig.setConfig(EtcdConfig.USER, yourUserName);
+        SentinelConfig.setConfig(EtcdConfig.PASSWORD, yourPassWord);
+        SentinelConfig.setConfig(EtcdConfig.CHARSET, "utf-8");
+        SentinelConfig.setConfig(EtcdConfig.AUTH_ENABLE, "true");
+
+        ReadableDataSource<String, List<FlowRule>> flowRuleEtcdDataSource = new EtcdDataSource<>(rule_key, (rule) -> JSON.parseArray(rule, FlowRule.class));
+        FlowRuleManager.register2Property(flowRuleEtcdDataSource.getProperty());
+        List<FlowRule> rules = FlowRuleManager.getRules();
+        System.out.println(rules);
+    }
+
+}

sentinel-extension/pom.xml

@@ -21,6 +21,7 @@
         <module>sentinel-parameter-flow-control</module>
         <module>sentinel-datasource-spring-cloud-config</module>
         <module>sentinel-datasource-consul</module>
+        <module>sentinel-datasource-etcd</module>
     </modules>
 
 </project>

sentinel-extension/sentinel-datasource-etcd/README.md

@@ -0,0 +1,40 @@
+# Sentinel DataSource Etcd
+
+Sentinel DataSource Etcd provides integration with Etcd so that Etcd
+can be the dynamic rule data source of Sentinel. The data source uses push model (watcher).
+
+To use Sentinel DataSource Etcd, you should add the following dependency:
+
+```xml
+<dependency>
+    <groupId>com.alibaba.csp</groupId>
+    <artifactId>sentinel-datasource-etcd</artifactId>
+    <version>x.y.z</version>
+</dependency>
+```
+Configure Etcd Connect Properties By Config File (for example sentinel.properties)
+
+```
+csp.sentinel.etcd.end.points=http://ip1:port1,http://ip2:port2
+csp.sentinel.etcd.user=your_user
+csp.sentinel.etcd.password=your_password
+csp.sentinel.etcd.charset=your_charset
+csp.sentinel.etcd.auth.enable=true //if ture open user/password or ssl check
+csp.sentinel.etcd.authority=authority //ssl
+```
+or JVM args(Add -D prefix)
+
+
+Then you can create an `EtcdDataSource` and register to rule managers.
+For instance:
+
+```java
+  //`rule_key` is the rule config key
+  ReadableDataSource<String, List<FlowRule>> flowRuleEtcdDataSource = new EtcdDataSource<>(rule_key, (rule) -> JSON.parseArray(rule, FlowRule.class));
+  FlowRuleManager.register2Property(flowRuleEtcdDataSource.getProperty());
+```
+
+> Note: It needs to update JDK version to JDK8
+
+
+We've also provided an example: [sentinel-demo-etcd-datasource](https://github.com/alibaba/Sentinel/tree/master/sentinel-demo/sentinel-demo-etcd-datasource)

sentinel-extension/sentinel-datasource-etcd/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-extension</artifactId>
+        <groupId>com.alibaba.csp</groupId>
+        <version>1.7.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>sentinel-datasource-etcd</artifactId>
+    <packaging>jar</packaging>
+
+    <properties>
+        <maven.compiler.source>1.8</maven.compiler.source>
+        <maven.compiler.target>1.8</maven.compiler.target>
+        <jetcd.version>0.3.0</jetcd.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.alibaba.csp</groupId>
+            <artifactId>sentinel-datasource-extension</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.etcd</groupId>
+            <artifactId>jetcd-core</artifactId>
+            <version>${jetcd.version}</version>
+        </dependency>
+
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>${maven.compiler.source}</source>
+                    <target>${maven.compiler.target}</target>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+
+
+</project>

sentinel-extension/sentinel-datasource-etcd/src/main/java/com/alibaba/csp/sentinel/datasource/etcd/EtcdConfig.java

@@ -0,0 +1,74 @@
+/*
+ * 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.datasource.etcd;
+
+import com.alibaba.csp.sentinel.config.SentinelConfig;
+import com.alibaba.csp.sentinel.util.StringUtil;
+
+/**
+ * Configure Etcd Connect Properties
+ *
+ * @author lianglin
+ * @since 1.7.0
+ */
+public final class EtcdConfig {
+
+
+    public final static String END_POINTS = "csp.sentinel.etcd.end.points";
+    public final static String USER = "csp.sentinel.etcd.user";
+    public final static String PASSWORD = "csp.sentinel.etcd.password";
+    public final static String CHARSET = "csp.sentinel.etcd.charset";
+    public final static String AUTH_ENABLE = "csp.sentinel.etcd.auth.enable";
+    public final static String AUTHORITY = "csp.sentinel.etcd.authority";
+
+    private final static String ENABLED = "true";
+
+
+    public static String getEndPoints() {
+        return SentinelConfig.getConfig(END_POINTS);
+    }
+
+    public static String getUser() {
+        return SentinelConfig.getConfig(USER);
+    }
+
+    public static String getPassword() {
+        return SentinelConfig.getConfig(PASSWORD);
+    }
+
+    public static String getCharset() {
+        String etcdCharSet = SentinelConfig.getConfig(CHARSET);
+        if (StringUtil.isNotBlank(etcdCharSet)) {
+            return etcdCharSet;
+        }
+        return SentinelConfig.charset();
+    }
+
+    public static boolean isAuthEnable() {
+        if (ENABLED.equalsIgnoreCase(SentinelConfig.getConfig(AUTH_ENABLE))) {
+            return true;
+        }
+        return false;
+    }
+
+    public static String getAuthority() {
+        return SentinelConfig.getConfig(AUTHORITY);
+    }
+
+    private EtcdConfig() {
+    };
+
+}

sentinel-extension/sentinel-datasource-etcd/src/main/java/com/alibaba/csp/sentinel/datasource/etcd/EtcdDataSource.java

@@ -0,0 +1,126 @@
+/*
+ * 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.datasource.etcd;
+
+import com.alibaba.csp.sentinel.datasource.AbstractDataSource;
+import com.alibaba.csp.sentinel.datasource.Converter;
+import com.alibaba.csp.sentinel.log.RecordLog;
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KeyValue;
+import io.etcd.jetcd.Watch;
+import io.etcd.jetcd.kv.GetResponse;
+import io.etcd.jetcd.watch.WatchEvent;
+
+import java.nio.charset.Charset;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * A read-only {@code DataSource} with Etcd backend. When the data in Etcd backend has been modified,
+ * Etcd will automatically push the new value so that the dynamic configuration can be real-time.
+ *
+ * @author lianglin
+ * @since 1.7.0
+ */
+public class EtcdDataSource<T> extends AbstractDataSource<String, T> {
+
+
+    private Client client;
+    private Watch.Watcher watcher;
+
+    private String key;
+    private Charset charset = Charset.forName(EtcdConfig.getCharset());
+
+    /**
+     * Create Etcd Data Source, Retrieve Connect Config Properties from ${@link EtcdConfig}
+     *
+     * @param key    Config key
+     * @param parser Value Parser
+     */
+    public EtcdDataSource(String key, Converter<String, T> parser) {
+        super(parser);
+        if (!EtcdConfig.isAuthEnable()) {
+            this.client = Client.builder()
+                    .endpoints(EtcdConfig.getEndPoints().split(",")).build();
+        } else {
+            this.client = Client.builder()
+                    .endpoints(EtcdConfig.getEndPoints().split(","))
+                    .user(ByteSequence.from(EtcdConfig.getUser(), charset))
+                    .password(ByteSequence.from(EtcdConfig.getPassword(), charset))
+                    .authority(EtcdConfig.getAuthority())
+                    .build();
+        }
+        this.key = key;
+        loadInitialConfig();
+        initWatcher();
+
+    }
+
+
+    private void loadInitialConfig() {
+        try {
+            T newValue = loadConfig();
+            if (newValue == null) {
+                RecordLog.warn("[EtcdDataSource] WARN: initial application is null, you may have to check your data source");
+            }
+            getProperty().updateValue(newValue);
+        } catch (Exception ex) {
+            RecordLog.warn("[EtcdDataSource] Error when loading initial application", ex);
+        }
+    }
+
+    private void initWatcher() {
+        watcher = client.getWatchClient().watch(ByteSequence.from(key, charset), (watchResponse) -> {
+            for (WatchEvent watchEvent : watchResponse.getEvents()) {
+                WatchEvent.EventType eventType = watchEvent.getEventType();
+                if (eventType == WatchEvent.EventType.PUT) {
+                    try {
+                        T newValue = loadConfig();
+                        getProperty().updateValue(newValue);
+                    } catch (Exception e) {
+                        RecordLog.warn("[EtcdDataSource] update rule config error: ", e);
+                    }
+                } else if (eventType == WatchEvent.EventType.DELETE) {
+                    getProperty().updateValue(null);
+                    RecordLog.info("[EtcdDataSource] clean rule config of {0}", key);
+                }
+            }
+        });
+    }
+
+
+    @Override
+    public String readSource() throws Exception {
+        CompletableFuture<GetResponse> responseFuture = client.getKVClient().get(ByteSequence.from(key, charset));
+        List<KeyValue> kvs = responseFuture.get().getKvs();
+        return kvs.size() == 0 ? null : kvs.get(0).getValue().toString(charset);
+    }
+
+    @Override
+    public void close() {
+        if (watcher != null) {
+            try {
+                watcher.close();
+            } catch (Exception ex) {
+                RecordLog.info("[EtcdDataSource] close watcher error", ex);
+            }
+        }
+        if (client != null) {
+            client.close();
+        }
+    }
+}

sentinel-extension/sentinel-datasource-etcd/src/test/java/com/alibaba/csp/sentinel/datasource/etcd/EtcdDataSourceTest.java

@@ -0,0 +1,123 @@
+/*
+ * 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.datasource.etcd;
+
+import com.alibaba.csp.sentinel.config.SentinelConfig;
+import com.alibaba.csp.sentinel.datasource.ReadableDataSource;
+import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
+import com.alibaba.csp.sentinel.slots.block.flow.FlowRuleManager;
+import com.alibaba.fastjson.JSON;
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KV;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+
+/**
+ * @author lianglin
+ * @since 1.7.0
+ */
+@Ignore(value = "Before run this test, you need to set up your etcd server.")
+public class EtcdDataSourceTest {
+
+
+    private final String endPoints = "http://127.0.0.1:2379";
+
+
+    @Before
+    public void setUp() {
+        SentinelConfig.setConfig(EtcdConfig.END_POINTS, endPoints);
+        FlowRuleManager.loadRules(new ArrayList<>());
+    }
+
+    @After
+    public void tearDown() {
+        SentinelConfig.setConfig(EtcdConfig.END_POINTS, "");
+        FlowRuleManager.loadRules(new ArrayList<>());
+    }
+
+    @Test
+    public void testReadSource() throws Exception {
+        EtcdDataSource dataSource = new EtcdDataSource("foo", value -> value);
+        KV kvClient = Client.builder()
+                .endpoints(endPoints)
+                .build().getKVClient();
+
+        kvClient.put(ByteSequence.from("foo".getBytes()), ByteSequence.from("test".getBytes()));
+        Assert.assertNotNull(dataSource.readSource().equals("test"));
+
+        kvClient.put(ByteSequence.from("foo".getBytes()), ByteSequence.from("test2".getBytes()));
+        Assert.assertNotNull(dataSource.getProperty().equals("test2"));
+    }
+
+    @Test
+    public void testDynamicUpdate() throws InterruptedException {
+        String demo_key = "etcd_demo_key";
+        ReadableDataSource<String, List<FlowRule>> flowRuleEtcdDataSource = new EtcdDataSource<>(demo_key, (value) -> JSON.parseArray(value, FlowRule.class));
+        FlowRuleManager.register2Property(flowRuleEtcdDataSource.getProperty());
+
+        KV kvClient = Client.builder()
+                .endpoints(endPoints)
+                .build().getKVClient();
+
+        final String rule1 = "[\n"
+                + "  {\n"
+                + "    \"resource\": \"TestResource\",\n"
+                + "    \"controlBehavior\": 0,\n"
+                + "    \"count\": 5.0,\n"
+                + "    \"grade\": 1,\n"
+                + "    \"limitApp\": \"default\",\n"
+                + "    \"strategy\": 0\n"
+                + "  }\n"
+                + "]";
+
+        kvClient.put(ByteSequence.from(demo_key.getBytes()), ByteSequence.from(rule1.getBytes()));
+        Thread.sleep(1000);
+
+        FlowRule flowRule = FlowRuleManager.getRules().get(0);
+        Assert.assertTrue(flowRule.getResource().equals("TestResource"));
+        Assert.assertTrue(flowRule.getCount() == 5.0);
+        Assert.assertTrue(flowRule.getGrade() == 1);
+
+        final String rule2 = "[\n"
+                + "  {\n"
+                + "    \"resource\": \"TestResource\",\n"
+                + "    \"controlBehavior\": 0,\n"
+                + "    \"count\": 6.0,\n"
+                + "    \"grade\": 3,\n"
+                + "    \"limitApp\": \"default\",\n"
+                + "    \"strategy\": 0\n"
+                + "  }\n"
+                + "]";
+
+        kvClient.put(ByteSequence.from(demo_key.getBytes()), ByteSequence.from(rule2.getBytes()));
+        Thread.sleep(1000);
+
+        flowRule = FlowRuleManager.getRules().get(0);
+        Assert.assertTrue(flowRule.getResource().equals("TestResource"));
+        Assert.assertTrue(flowRule.getCount() == 6.0);
+        Assert.assertTrue(flowRule.getGrade() == 3);
+
+
+    }
+}