Es driver (#59)

* intial es driver framework. * Add unit test.
openmessaging · Nov 8, 2022 · f3f4177 · f3f4177
1 parent 95142ab
commit f3f4177
Show file tree

Hide file tree

Showing 16 changed files with 783 additions and 0 deletions.
diff --git a/chaos-framework/pom.xml b/chaos-framework/pom.xml
@@ -100,6 +100,11 @@
             <groupId>com.aliyun.oss</groupId>
             <artifactId>aliyun-sdk-oss</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.openmessaging</groupId>
+            <artifactId>driver-elasticsearch</artifactId>
+            <version>0.6.5-SNAPSHOT</version>
+        </dependency>
     </dependencies>
 
     <build>

diff --git a/driver-elasticsearch/elasticsearch.yaml b/driver-elasticsearch/elasticsearch.yaml
@@ -0,0 +1,34 @@
+#
+# 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.
+#
+
+name: ElasticSearch
+driverClass: io.openchaos.driver.elasticsearch.ElasticSearchDriver
+
+endToEndLatencyCheck: false
+
+# ElasticSearch cluster configuration
+
+# Nodes for ElasticSearch-Server
+nodes:
+  - 192.144.239.212 # replace with ip or domain name, such as 192.168.0.2
+
+# ElasticSearch configuration
+elasticsearchVersion: 8.10
+installDir: es-chaos-test # you could set existent location for ElasticSearch
+isSsl: false
+
+# ElasticSearch Client configuration
+username: elastic
+password: elastic
+port: 9200
diff --git a/driver-elasticsearch/pom.xml b/driver-elasticsearch/pom.xml
@@ -0,0 +1,92 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<!--
+
+    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.
+
+-->
+
+<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-elasticsearch</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>driver-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <version>${jackson-databind.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-yaml</artifactId>
+            <version>${jackson-dataformat-yaml.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+            <version>${commons-lang3.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <version>${mockito.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.elasticsearch.client</groupId>
+            <artifactId>elasticsearch-rest-client</artifactId>
+            <version>${es-rest-client.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpclient</artifactId>
+            <version>${httpclient.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpcore</artifactId>
+            <version>${httpcore.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+        </dependency>
+    </dependencies>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+        <jackson-databind.version>2.13.2.2</jackson-databind.version>
+        <jackson-dataformat-yaml.version>2.10.2</jackson-dataformat-yaml.version>
+        <commons-lang3.version>3.8.1</commons-lang3.version>
+        <mockito.version>4.6.1</mockito.version>
+        <es-rest-client.version>8.0.0</es-rest-client.version>
+        <httpclient.version>4.5.13</httpclient.version>
+        <httpcore.version>4.4.15</httpcore.version>
+    </properties>
+
+</project>
diff --git a/...asticsearch/src/main/java/io/openchaos/driver/elasticsearch/ElasticSearchChaosClient.java b/...asticsearch/src/main/java/io/openchaos/driver/elasticsearch/ElasticSearchChaosClient.java
@@ -0,0 +1,134 @@
+package io.openchaos.driver.elasticsearch;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.openchaos.common.InvokeResult;
+import io.openchaos.driver.kv.KVClient;
+import io.openchaos.driver.elasticsearch.core.Document;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpStatus;
+import org.apache.http.entity.ContentType;
+import org.apache.http.nio.entity.NStringEntity;
+import org.apache.http.util.EntityUtils;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.RestClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.*;
+
+public class ElasticSearchChaosClient implements KVClient {
+    private static final Logger log = LoggerFactory.getLogger(ElasticSearchChaosClient.class);
+    private RestClient esClient;
+    private final String endpoint = "openchaos";
+    private static final ObjectMapper objectMapper = new ObjectMapper().configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+    private final String KEY = "openchaosTest";
+
+    public ElasticSearchChaosClient(RestClient client) {
+        esClient = client;
+    }
+
+    @Override
+    public void start() {
+    }
+
+    @Override
+    public void close() {
+    }
+
+    @Override
+    public InvokeResult put(Optional<String> key, String value) {
+        try {
+            String id = "/" + KEY + value;
+            String method = "POST";
+            Request request = new Request(method, "/" + endpoint + "/_create" + id);
+            String jsonStr = serialize(key, value);
+            NStringEntity entity = new NStringEntity(jsonStr, ContentType.APPLICATION_JSON);
+            request.setEntity(entity);
+            Response response = esClient.performRequest(request);
+            if (response.getStatusLine().getStatusCode() != HttpStatus.SC_CREATED) {
+                log.error("Method failed: " + response.getStatusLine());
+            } else {
+                return InvokeResult.SUCCESS;
+            }
+        } catch (IOException e) {
+            log.error(e.getMessage());
+            e.printStackTrace();
+        }
+        return InvokeResult.FAILURE;
+    }
+
+    @Override
+    public List<String> getAll(Optional<String> key, int putInvokeCount) {
+        List<String> values = new ArrayList<>();
+        for (int i = 0; i < putInvokeCount; ++i) {
+            Document document = fullTextMatch(key, i);
+            values.add(document.getValue());
+        }
+        return values;
+    }
+
+    @Override
+    public List<String> getAll(Optional<String> key) {
+        return null;
+    }
+
+    private String serialize(Optional<String> key, String value) throws JsonProcessingException {
+        Document document = new Document(String.valueOf(key), value);
+        document.setKey(String.valueOf(key));
+        document.setValue(value);
+        return objectMapper.writeValueAsString(document);
+    }
+
+    private List<Document> deserialize(HttpEntity entity) throws IOException {
+        String responseBody = EntityUtils.toString(entity);
+        JSONObject resultObject = JSON.parseObject(responseBody);
+        JSONObject hitsObject = JSON.parseObject(resultObject.get("hits").toString());
+        JSONArray jsonArray = JSON.parseArray(hitsObject.get("hits").toString());
+
+        List<Document> documents = new ArrayList<>();
+        for (Object object : jsonArray) {
+            JSONObject jsonObject1 = JSON.parseObject(object.toString());
+            JSONObject jsonObject2 = JSON.parseObject(jsonObject1.get("_source").toString());
+            Document document = new Document(String.valueOf(jsonObject2.get("key")), String.valueOf(jsonObject2.get("value")));
+            documents.add(document);
+        }
+        return documents;
+    }
+
+    /**
+     *    Full text match.
+     */
+    private Document fullTextMatch(Optional<String> key, int i) {
+        Map<String, Object> jsonMap = new HashMap<>();
+
+        Map<String, Object> matchMap = new HashMap<>();
+        Map<String, String> fieldsMap = new HashMap<>();
+        fieldsMap.put("value", String.valueOf(i));
+        matchMap.put("match", fieldsMap);
+        jsonMap.put("query", matchMap);
+
+        JSONObject jsonObject = new JSONObject(jsonMap);
+        String jsonString = jsonObject.toString();
+
+        Request request = new Request("GET", "/" + endpoint + "/_search");
+        request.addParameter("pretty", "true");
+        request.setEntity(new NStringEntity(jsonString, ContentType.APPLICATION_JSON));
+        try {
+            Response response = esClient.performRequest(request);
+            return deserialize(response.getEntity()).get(0);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public void setEsClient(RestClient esClient) {
+        this.esClient = esClient;
+    }
+}
diff --git a/...elasticsearch/src/main/java/io/openchaos/driver/elasticsearch/ElasticSearchChaosNode.java b/...elasticsearch/src/main/java/io/openchaos/driver/elasticsearch/ElasticSearchChaosNode.java
@@ -0,0 +1,91 @@
+package io.openchaos.driver.elasticsearch;
+
+import io.openchaos.common.utils.KillProcessUtil;
+import io.openchaos.common.utils.PauseProcessUtil;
+import io.openchaos.common.utils.SshUtil;
+import io.openchaos.driver.elasticsearch.config.ElasticSearchConfig;
+import io.openchaos.driver.kv.KVNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+public class ElasticSearchChaosNode implements KVNode {
+    private static final Logger log = LoggerFactory.getLogger(ElasticSearchChaosNode.class);
+    private static final String ES_PROCESS_NAME = "elasticsearch";
+    private String installDir = "es-chaos-test";
+    private String node;
+    private List<String> nodes;
+    private String elasticsearchVersion = "8.10";
+
+    public ElasticSearchChaosNode(String node, List<String> nodes, ElasticSearchConfig elasticSearchConfig) {
+        this.node = node;
+        this.nodes = nodes;
+        if (elasticSearchConfig.installDir != null && !elasticSearchConfig.installDir.isEmpty()) {
+            this.installDir = elasticSearchConfig.installDir;
+        }
+        if (elasticSearchConfig.elasticsearchVersion != null && !elasticSearchConfig.elasticsearchVersion.isEmpty()) {
+            this.elasticsearchVersion = elasticSearchConfig.elasticsearchVersion;
+        }
+    }
+    @Override
+    public void setup() {
+        //todo 安装集群
+    }
+
+    @Override
+    public void teardown() {
+        stop();
+    }
+
+    @Override
+    public void start() {
+        try {
+            SshUtil.execCommandInDir(node, installDir, "./bin/elasticsearch -d");
+        } catch (Exception e) {
+            log.error("Node {} start es node failed", node, e);
+            throw new RuntimeException(e);
+        }
+
+    }
+
+    @Override
+    public void stop() {
+        try {
+            KillProcessUtil.kill(node, ES_PROCESS_NAME);
+        } catch (Exception e) {
+            log.error("Node {} stop es node failed", node, e);
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public void kill() {
+        try {
+            KillProcessUtil.forceKill(node, ES_PROCESS_NAME);
+        } catch (Exception e) {
+            log.error("Node {} kill es node failed", node, e);
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public void pause() {
+        try {
+            PauseProcessUtil.suspend(node, ES_PROCESS_NAME);
+        } catch (Exception e) {
+            log.error("Node {} pause es node failed", node, e);
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public void resume() {
+        try {
+            PauseProcessUtil.resume(node, ES_PROCESS_NAME);
+        } catch (Exception e) {
+            log.error("Node {} resume es node failed", node, e);
+            throw new RuntimeException(e);
+        }
+    }
+}