add redis implement of datasource

sentinel-extension/pom.xml

@@ -16,6 +16,7 @@
         <module>sentinel-datasource-nacos</module>
         <module>sentinel-datasource-zookeeper</module>
         <module>sentinel-datasource-apollo</module>
+        <module>sentinel-datasource-redis</module>
         <module>sentinel-annotation-aspectj</module>
     </modules>
 

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

@@ -0,0 +1,37 @@
+# Sentinel DataSource Redis
+
+Sentinel DataSource Redis provides integration with Redis so that Redis
+can be the dynamic rule data source of Sentinel. The data source uses push model (listener) with redis pub/sub feature.
+
+this RedisDataSource implement only Redis Standalone. if you want to use redis cluster. please read this [Redis Cluster PUB/SUB](https://github.com/lettuce-io/lettuce-core/wiki/Pub-Sub),
+ then you can implement by yourself.
+
+To use Sentinel DataSource Redis, you should add the following dependency:
+
+```xml
+<dependency>
+    <groupId>com.alibaba.csp</groupId>
+    <artifactId>sentinel-datasource-redis</artifactId>
+    <version>x.y.z</version>
+</dependency>
+
+```
+
+Then you can create an `RedisDataSource` and register to rule managers.
+For instance:
+
+```java
+
+DataSource<String, List<FlowRule>> redisDataSource = new RedisDataSource<List<FlowRule>>(client, ruleKey, channel, flowConfigParser);
+FlowRuleManager.register2Property(redisDataSource.getProperty());
+```
+
+*client* : we use [lettuce](https://lettuce.io/) as redis-cli client. you can build client this way [how client build](https://github.com/lettuce-io/lettuce-core/wiki/Basic-usage) . 
+
+*ruleKey* : when the json rule data publish. it also should save to the key for init read.
+
+*channel* : the channel to listen.  
+
+you can also create multi data source listen for different rule type. 
+
+you can run test case for usage.

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

@@ -0,0 +1,55 @@
+<?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.2.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>sentinel-datasource-redis</artifactId>
+    <packaging>jar</packaging>
+
+    <properties>
+        <zookeeper.version>3.4.13</zookeeper.version>
+        <curator.version>4.0.1</curator.version>
+        <curator.test.version>2.12.0</curator.test.version>
+        <lettuce.version>5.0.1.RELEASE</lettuce.version>
+        <redis.mock.version>0.1.6</redis.mock.version>
+    </properties>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>io.lettuce</groupId>
+            <artifactId>lettuce-core</artifactId>
+            <version>${lettuce.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.alibaba.csp</groupId>
+            <artifactId>sentinel-datasource-extension</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>ai.grakn</groupId>
+            <artifactId>redis-mock</artifactId>
+            <version>${redis.mock.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+</project>

sentinel-extension/sentinel-datasource-redis/src/main/java/com/alibaba/csp/sentinel/datasource/redis/RedisDataSource.java

@@ -0,0 +1,82 @@
+package com.alibaba.csp.sentinel.datasource.redis;
+
+import com.alibaba.csp.sentinel.datasource.AbstractDataSource;
+import com.alibaba.csp.sentinel.datasource.ConfigParser;
+import com.alibaba.csp.sentinel.log.RecordLog;
+import io.lettuce.core.RedisClient;
+import io.lettuce.core.api.sync.RedisCommands;
+import io.lettuce.core.pubsub.RedisPubSubAdapter;
+import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
+import io.lettuce.core.pubsub.api.sync.RedisPubSubCommands;
+
+
+/**
+ * Redis data resource. when data publish in redis,this will update rule config in time
+ *
+ * @author moon tiger
+ */
+
+public class RedisDataSource<T> extends AbstractDataSource<String, T> {
+
+    private RedisClient redisClient = null;
+
+    private String ruleKey;
+
+    public RedisDataSource(RedisClient client, String ruleKey, String channel, ConfigParser<String, T> parser) {
+        super(parser);
+        this.redisClient = client;
+        this.ruleKey = ruleKey;
+        initConfig();
+        subscribeFromChannel(channel);
+    }
+
+    private void subscribeFromChannel(String channel) {
+        StatefulRedisPubSubConnection<String, String> pubSubConnection = redisClient.connectPubSub();
+        RedisPubSubAdapter<String, String> adapterListener = new DelegatingRedisPubSubListener();
+        pubSubConnection.addListener(adapterListener);
+        RedisPubSubCommands<String, String> sync = pubSubConnection.sync();
+        sync.subscribe(channel);
+    }
+
+
+    private void initConfig() {
+        try {
+            loadConfig();
+        } catch (Exception e) {
+            RecordLog.info("[RedisDataSource] Error when loading initial config", e);
+        }
+    }
+
+    @Override
+    public String readSource() throws Exception {
+        if (this.redisClient == null) {
+            throw new IllegalStateException("redis client has not been initialized or error occurred");
+        }
+        RedisCommands<String, String> stringRedisCommands = redisClient.connect().sync();
+        String value = stringRedisCommands.get(ruleKey);
+        if (value != null) {
+            return value;
+        }
+        RecordLog.warn("rules is empty in redis. key:" + ruleKey);
+        return null;
+    }
+
+    @Override
+    public void close() throws Exception {
+        redisClient.shutdown();
+    }
+
+
+    private class DelegatingRedisPubSubListener extends RedisPubSubAdapter<String, String> {
+
+        DelegatingRedisPubSubListener() {
+        }
+
+        @Override
+        public void message(String channel, String message) {
+            getProperty().updateValue(parser.parse(message));
+        }
+    }
+
+
+}

sentinel-extension/sentinel-datasource-redis/src/test/java/com/alibaba/csp/sentinel/datasource/redis/RedisDataSourceTest.java

@@ -0,0 +1,111 @@
+package com.alibaba.csp.sentinel.datasource.redis;
+
+import ai.grakn.redismock.RedisServer;
+import com.alibaba.csp.sentinel.datasource.ConfigParser;
+import com.alibaba.csp.sentinel.datasource.DataSource;
+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 io.lettuce.core.RedisClient;
+import io.lettuce.core.RedisURI;
+import io.lettuce.core.api.sync.RedisCommands;
+import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.List;
+
+public class RedisDataSourceTest {
+
+    private static RedisServer server = null;
+
+    private RedisClient client;
+
+    private String ruleKey = "sentinel.flow.rulekey";
+
+    private String channel = "sentinel.flow.channel";
+
+    @Before
+    public void buildResource() {
+        try {
+            //bind to a random port
+            server = RedisServer.newRedisServer();
+            server.start();
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+        ConfigParser<String, List<FlowRule>> flowConfigParser = buildFlowConfigParser();
+        client = RedisClient.create(RedisURI.create(server.getHost(), server.getBindPort()));
+        initRedisRuleData();
+        DataSource<String, List<FlowRule>> redisDataSource = new RedisDataSource<List<FlowRule>>(client, ruleKey, channel, flowConfigParser);
+        FlowRuleManager.register2Property(redisDataSource.getProperty());
+    }
+
+    @Test
+    public void pub_msg_and_receive_success() {
+        List<FlowRule> rules = FlowRuleManager.getRules();
+        Assert.assertTrue(rules.size() == 0);
+        int maxQueueingTimeMs = 480;
+        StatefulRedisPubSubConnection<String, String> connection = client.connectPubSub();
+        String flowRules = "[{\"resource\":\"test\", \"limitApp\":\"default\", \"grade\":1, \"count\":\"0.0\", \"strategy\":0, \"refResource\":null, " +
+                "\"controlBehavior\":0, \"warmUpPeriodSec\":10, \"maxQueueingTimeMs\":" + maxQueueingTimeMs + ", \"controller\":null}]";
+        connection.sync().publish(channel, flowRules);
+        try {
+            Thread.sleep(2000);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+        rules = FlowRuleManager.getRules();
+        Assert.assertTrue(rules.get(0).getMaxQueueingTimeMs() == maxQueueingTimeMs);
+    }
+
+
+    @Test
+    public void init_and_parse_flow_rule_success() {
+        RedisCommands<String, String> stringRedisCommands = client.connect().sync();
+        String value = stringRedisCommands.get(ruleKey);
+        List<FlowRule> flowRules = buildFlowConfigParser().parse(value);
+        Assert.assertTrue(flowRules.size() == 1);
+        stringRedisCommands.del(ruleKey);
+    }
+
+    @Test
+    public void read_resource_fail() {
+        RedisCommands<String, String> stringRedisCommands = client.connect().sync();
+        stringRedisCommands.del(ruleKey);
+        String value = stringRedisCommands.get(ruleKey);
+        Assert.assertTrue(value == null);
+    }
+
+
+    @After
+    public void clearResource() {
+        RedisCommands<String, String> stringRedisCommands = client.connect().sync();
+        stringRedisCommands.del(ruleKey);
+        client.shutdown();
+        server.stop();
+        server = null;
+    }
+
+    private ConfigParser<String, List<FlowRule>> buildFlowConfigParser() {
+        return new ConfigParser<String, List<FlowRule>>() {
+            @Override
+            public List<FlowRule> parse(String source) {
+                return JSON.parseObject(source, new TypeReference<List<FlowRule>>() {
+                });
+            }
+        };
+    }
+
+    private void initRedisRuleData() {
+        String flowRulesJson = "[{\"resource\":\"test\", \"limitApp\":\"default\", \"grade\":1, \"count\":\"0.0\", \"strategy\":0, \"refResource\":null, " +
+                "\"controlBehavior\":0, \"warmUpPeriodSec\":10, \"maxQueueingTimeMs\":500, \"controller\":null}]";
+        RedisCommands<String, String> stringRedisCommands = client.connect().sync();
+        String ok = stringRedisCommands.set(ruleKey, flowRulesJson);
+        Assert.assertTrue(ok.equals("OK"));
+    }
+}