[ISSUE openmessaging#46] Add support for etcd, and put forward the me…

…thod of closing resources in ChaosState interface. (openmessaging#47) * add etcd etcd implementation * Update pom.xml update version * Update EtcdClientConfig.java * Update EtcdConfig.java * Update EtcdChaosClient.java * Update EtcdChaosDriver.java * Update EtcdNode.java * Update EtcdState.java * Update pom.xml * Update EtcdChaosClient.java * Update EtcdNode.java * Update ChaosState.java * Update ChaosState.java
codelipenghui · Jan 4, 2022 · 467a4c9 · 467a4c9
1 parent 151a300
commit 467a4c9
Show file tree

Hide file tree

Showing 15 changed files with 591 additions and 10 deletions.
diff --git a/chaos-common/src/main/java/io/openchaos/common/utils/ServiceUtil.java b/chaos-common/src/main/java/io/openchaos/common/utils/ServiceUtil.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 io.openchaos.common.utils;
+
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ServiceUtil {
+    private static final String STOP_ACTION = "stop";
+    private static final String START_ACTION = "start";
+    private static final String RESTART_ACTION = "restart";
+
+    private static final Logger log = LoggerFactory.getLogger(ServiceUtil.class);
+
+    public static void stop(String node, String serviceName) throws Exception {
+        log.info("Start stop node {} service {}.", node, serviceName);
+        action(node, serviceName, STOP_ACTION);
+    }
+
+    public static void start(String node, String serviceName) throws Exception {
+        log.info("Start start node {} service {}.", node, serviceName);
+        action(node, serviceName, START_ACTION);
+    }
+
+    public static void restart(String node, String serviceName) throws Exception {
+        log.info("Start restart node {} service {}.", node, serviceName);
+        action(node, serviceName, RESTART_ACTION);
+    }
+
+    private static void action(String node, String serviceName, String action) throws Exception {
+        SshUtil.execCommand(node, String.format("service %s %s", serviceName, action));
+    }
+}
diff --git a/chaos-framework/pom.xml b/chaos-framework/pom.xml
@@ -84,6 +84,10 @@
             <groupId>${project.groupId}</groupId>
             <artifactId>driver-redis</artifactId>
         </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>driver-etcd</artifactId>
+        </dependency>
         <dependency>
             <groupId>com.panayotis</groupId>
             <artifactId>javaplot</artifactId>

diff --git a/chaos-framework/src/main/java/io/openchaos/generator/FaultGenerator.java b/chaos-framework/src/main/java/io/openchaos/generator/FaultGenerator.java
@@ -155,11 +155,14 @@ private static List<FaultOperation> leaderPartition(Collection<String> nodes, St
         }
         chaosState.initialize(clusterName, addr);
         Set<String> leaderNode = null;
+        Set<String> leaderAddr = new HashSet<>();
         leaderNode = chaosState.getLeader();
-
+        for (String leader : leaderNode) {
+            leaderAddr.add(leader.substring(0, leader.indexOf(":")));
+        }
         List<String> shuffleNodes = new ArrayList<>(nodes);
         Collections.shuffle(shuffleNodes);
-        Set<String> partition1 = new HashSet<>(leaderNode);
+        Set<String> partition1 = new HashSet<>(leaderAddr);
         Set<String> partition2 = new HashSet<>(nodes);
         int i = 0;
         while (partition1.size() < num) {
@@ -168,7 +171,7 @@ private static List<FaultOperation> leaderPartition(Collection<String> nodes, St
         }
         partition2.removeAll(partition1);
         partition1.forEach(node -> operations.add(getPartitionOperation(faultName, node, partition2)));
-
+        chaosState.close();
         return operations;
     }
 
@@ -267,12 +270,20 @@ private static List<FaultOperation> faultInRandomNumberLeaderNodes(String faultN
         chaosState.initialize(metaName, metaNode);
         Set<String> leaderNode;
         leaderNode = chaosState.getLeader();
-        List<String> shuffleNodes = new ArrayList<>(leaderNode);
+
+        Set<String> leaderAddr = new HashSet<>();
+        for (String leader : leaderNode) {
+            leaderAddr.add(leader.substring(0, leader.indexOf(":")));
+        }
+
+        List<String> shuffleNodes = new ArrayList<>(leaderAddr);
         Collections.shuffle(shuffleNodes);
+
         for (int i = 0; i < num; i++) {
             FaultOperation operation = new FaultOperation(faultName, shuffleNodes.get(i));
             operations.add(operation);
         }
+        chaosState.close();
         return operations;
     }
 

diff --git a/driver-api/src/main/java/io/openchaos/driver/ChaosState.java b/driver-api/src/main/java/io/openchaos/driver/ChaosState.java
@@ -19,5 +19,10 @@ public interface ChaosState {
 
     void initialize(String metaName, String metaNode);
 
+    /**
+     * Returns the leader nodes of the cluster. The format is as follows: ip:port
+     */
     Set<String> getLeader();
+
+    void close();
 }
diff --git a/driver-etcd/etcd.yaml b/driver-etcd/etcd.yaml
@@ -0,0 +1,45 @@
+  <!--
+
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you 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.
+
+    -->
+
+name: Etcd
+driverClass: io.openchaos.driver.etcd.EtcdChaosDriver
+
+# ETCD cluster configuration
+nodes:
+  - n1 # replace with ip or domain name, such as 192.168.0.1
+  - n2
+  - n3
+
+# ETCD configuration
+version: v3
+https: false
+auth: true
+endpoints: http://192.168.0.1:2379,http://192.168.0.2:2379,http://192.168.0.3:2379
+
+# ETCD Client configuration
+username:
+password:
+keepAlive:
+healthCheck:
+retryCount:
+connectTimeout:  # ms
+sslMode:
+readTimeout:
diff --git a/driver-etcd/pom.xml b/driver-etcd/pom.xml
@@ -0,0 +1,53 @@
+<?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>openchaos</artifactId>
+        <groupId>io.openmessaging</groupId>
+        <version>0.6.5-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>driver-etcd</artifactId>
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+
+    <properties>
+        <jetcd.version>0.5.11</jetcd.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>driver-api</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.etcd</groupId>
+            <artifactId>jetcd-core</artifactId>
+            <version>${jetcd.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-yaml</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <version>1.18.20</version>
+        </dependency>
+    </dependencies>
+
+</project>
diff --git a/driver-etcd/src/main/java/io/openchaos/driver/etcd/EtcdChaosClient.java b/driver-etcd/src/main/java/io/openchaos/driver/etcd/EtcdChaosClient.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements.  See the NOTICE
+ * file distributed with this work for additional information regarding copyright ownership.  The ASF licenses this file
+ * to you 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 io.openchaos.driver.etcd;
+
+import com.google.common.base.Charsets;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KV;
+import io.etcd.jetcd.KeyValue;
+import io.etcd.jetcd.kv.GetResponse;
+import io.etcd.jetcd.kv.PutResponse;
+import io.openchaos.common.InvokeResult;
+import io.openchaos.driver.kv.KVClient;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+
+@Slf4j
+public class EtcdChaosClient implements KVClient {
+
+    private final Client client;
+
+    public EtcdChaosClient(Client client) {
+        this.client = client;
+    }
+
+    public Client getClient() {
+        return client;
+    }
+
+    @Override
+    public void start() {
+
+    }
+
+    @Override
+    public void close() {
+        Optional.ofNullable(client).ifPresent(c -> c.close());
+        log.info("Close etcd client");
+    }
+
+    @Override
+    public InvokeResult put(Optional<String> key, String value) {
+        try {
+            PutResponse response = client.getKVClient()
+                .put(ByteSequence.from(key.get() + value, Charsets.UTF_8), ByteSequence.from(value, Charsets.UTF_8))
+                .get();
+        } catch (InterruptedException | ExecutionException e) {
+            log.error("Etcd put failed.", e);
+            return InvokeResult.FAILURE;
+        }
+        return InvokeResult.SUCCESS;
+    }
+
+    @Override
+    public List<String> getAll(Optional<String> key, int putInvokeCount) {
+        List<String> results = new LinkedList<>();
+        KV kv = client.getKVClient();
+
+        try {
+            for (int i = 0; i < putInvokeCount; i++) {
+                GetResponse response = kv.get(ByteSequence.from(key.get() + i, Charsets.UTF_8)).get();
+                if (response.getKvs().isEmpty()) {
+                    continue;
+                }
+                for (KeyValue keyValue : response.getKvs()) {
+                    results.add(keyValue.getValue().toString(Charsets.UTF_8));
+                }
+            }
+        } catch (InterruptedException | ExecutionException e) {
+            log.error("Get etcd key failed.", e);
+        }
+        return results;
+    }
+
+    @Override
+    public List<String> getAll(Optional<String> key) {
+        return getAll(key, 1);
+    }
+}