Repository: lf-edge/ekuiper
Pull request #1136 (2021-12-28 07:40:09)

fix(ui): backend support for ui update config key
Signed-off-by: Jianxiang Ran <rxan_embedded@163.com>


Patch:
diff --git a/docs/en_US/cli/tables.md b/docs/en_US/cli/tables.md
index 3a8e6ec9..fab78b7b 100644
--- a/docs/en_US/cli/tables.md
+++ b/docs/en_US/cli/tables.md
@@ -74,6 +74,7 @@ FORMAT: json
 KEY: id
 DATASOURCE: lookup.json
 ```
+ *Note*: eKuiper do not support query table data by cli. Users need join the table with a stream and check the result
 
 ## drop a table
 
diff --git a/docs/en_US/rules/sources/file.md b/docs/en_US/rules/sources/file.md
index d40f9da6..652fbe9d 100644
--- a/docs/en_US/rules/sources/file.md
+++ b/docs/en_US/rules/sources/file.md
@@ -3,13 +3,14 @@
 eKuiper provides built-in support for reading file content into the eKuiper processing pipeline. The file source is usually used as a [table](../../sqls/tables.md) and it is the default type for create table statement.
 
 ```sql
-CREATE TABLE table1 (
+create table table1 (
     name STRING,
     size BIGINT,
     id BIGINT
 ) WITH (DATASOURCE="lookup.json", FORMAT="json", TYPE="file");
 ```
 
+You can use [cli](../../cli/tables.md) or [rest api](../../restapi/tables.md) to manage the tables.
 
 The configure file for the file source is in */etc/sources/file.yaml* in which the path to the file can be specified.
 
diff --git a/docs/zh_CN/cli/tables.md b/docs/zh_CN/cli/tables.md
new file mode 100644
index 00000000..b259ad38
--- /dev/null
+++ b/docs/zh_CN/cli/tables.md
@@ -0,0 +1,77 @@
+# 表管理
+eKuiper 表命令行工具允许您管理表，例如创建、描述、显示和删除。
+
+## 创建表
+该命令用于创建表。更详细的表定义信息请参考[tables](../sqls/tables.md)。
+```shell
+create table $table_name $table_def | create table -f $table_def_file
+```
+
+- 在命令行中定义表信息.
+  以下例子通过命令行创建了一个名为 ``my_table``的表。
+```shell
+# bin/kuiper create table my_table '(id bigint, name string, score float) WITH ( datasource = "lookup.json", FORMAT = "json", KEY = "id");'
+table my_table created
+```
+
+
+- 在文件中指定表定义。如果表格很复杂，或者表格已经写在具有良好组织格式的文本文件中，您可以通过“-f”选项指定表格定义。
+  例如:
+```shell
+# bin/kuiper create table -f /tmp/my_table.txt
+table my_table created
+```
+
+  以下是 ``my_table.txt``的内容.
+```
+my_table(id bigint, name string, score float)
+    WITH ( datasource = "lookup.json", FORMAT = "json", KEY = "id");
+```
+
+## 查看所有的表 
+该命令用于显示 eKuiper 中定义的所有表。
+```shell
+show tables
+```
+
+例如:
+```shell
+# bin/kuiper show tables
+my_table
+```
+
+## 查看表的详细信息
+
+该命令用于打印表的详细定义。
+```shell
+describe table $table_name
+```
+
+例如:
+```shell
+# bin/kuiper describe table my_table
+Fields
+--------------------------------------------------------------------------------
+id	bigint
+name	string
+score	float
+
+FORMAT: json
+KEY: id
+DATASOURCE: lookup.json
+```
+
+*注意*： 目前 eKuiper 不支持查看表的内容，用户可直接将表与流 join 查看结果。
+
+## 删除表
+
+该命令用于删除表
+```shell
+drop table $table_name
+```
+
+例如:
+```shell
+# bin/kuiper drop table my_table
+table my_table dropped
+```
\ No newline at end of file
diff --git a/docs/zh_CN/restapi/streams.md b/docs/zh_CN/restapi/streams.md
index 8db06c02..a0154bfc 100644
--- a/docs/zh_CN/restapi/streams.md
+++ b/docs/zh_CN/restapi/streams.md
@@ -1,3 +1,4 @@
+
 # 流管理
 
 eKuiper REST api 可用于管理流，例如创建、描述、显示和删除流定义。
diff --git a/docs/zh_CN/restapi/tables.md b/docs/zh_CN/restapi/tables.md
new file mode 100644
index 00000000..96b21e7c
--- /dev/null
+++ b/docs/zh_CN/restapi/tables.md
@@ -0,0 +1,81 @@
+# 表管理
+
+eKuiper 支持 REST api 方式管理表，例如创建、描述、显示和删除。
+
+## 创建表
+
+该 API 用于创建表。更详细的表定义信息请参考[tables](../sqls/tables.md)。
+```shell
+POST http://localhost:9081/tables
+```
+
+请求示例，请求是一个带有 `sql` 字段的 json 字符串。
+```json
+{"sql":"create table my_table (id bigint, name string, score float) WITH ( datasource = \"lookup.json\", FORMAT = \"json\", KEY = \"id\")"}
+```
+这个API可以运行任何表的sql语句，不仅仅是建表。
+
+## 查看所有的表
+
+此API 用于显示 eKuiper 中定义的所有表
+```shell
+GET http://localhost:9081/tables
+```
+
+返回示例:
+```json
+["mytable"]
+```
+
+## 查看表的详细信息
+
+该 API 用于打印表的详细定义。
+```shell
+GET http://localhost:9081/tables/{id}}
+```
+
+返回示例:
+```shell
+{
+  "Name": "demo",
+  "StreamFields": [
+    {
+      "Name": "temperature",
+      "FieldType": {
+        "Type": 2
+      }
+    },
+    {
+      "Name": "ts",
+      "FieldType": {
+        "Type": 1
+      }
+    }
+  ],
+  "Options": {
+    "DATASOURCE": "lookup.json",
+    "FORMAT": "JSON"
+  }
+}
+```
+
+## 更新表
+
+该 API 用于更新表的定义。
+```shell
+PUT http://localhost:9081/tables/{id}
+```
+
+路径参数 `id` 是旧表的 id 或名称。
+请求示例，请求是一个带有 `sql` 字段的 json 字符串。
+```json
+{"sql":"create table my_table (id bigint, name string, score float) WITH ( datasource = \"topic/temperature\", FORMAT = \"json\", KEY = \"id\")"}
+```
+
+## 删除表
+
+该 API 用于删除表。
+```shell
+DELETE http://localhost:9081/tables/{id}
+```
+
diff --git a/docs/zh_CN/rules/sources/file.md b/docs/zh_CN/rules/sources/file.md
index 55e7bb6f..079ad578 100644
--- a/docs/zh_CN/rules/sources/file.md
+++ b/docs/zh_CN/rules/sources/file.md
@@ -10,6 +10,8 @@ CREATE TABLE table1 (
 ) WITH (DATASOURCE="lookup.json", FORMAT="json", TYPE="file");
 ```
 
+您可以使用 [cli](../../cli/tables.md) 或 [rest api](../../restapi/tables.md) 来管理表
+
 文件源的配置文件是 */etc/sources/file.yaml* ，可以在其中指定文件的路径。
 
 ```yaml
diff --git a/internal/meta/yamlConfigMeta.go b/internal/meta/yamlConfigMeta.go
index f9037d12..aa0cd4e1 100644
--- a/internal/meta/yamlConfigMeta.go
+++ b/internal/meta/yamlConfigMeta.go
@@ -139,7 +139,7 @@ func AddSourceConfKey(plgName, confKey, language string, content []byte) error {
 	}
 
 	if err := cfgOps.AddConfKey(confKey, reqField); err != nil {
-		return fmt.Errorf(`%s%s`, getMsg(language, source, "confkey_already_exist"), configOperatorKey)
+		return err
 	}
 
 	err = cfgOps.SaveCfgToFile()
@@ -177,87 +177,12 @@ func AddConnectionConfKey(plgName, confKey, language string, content []byte) err
 	}
 
 	if err := cfgOps.AddConfKey(confKey, reqField); err != nil {
-		return fmt.Errorf(`%s%s`, getMsg(language, source, "confkey_already_exist"), configOperatorKey)
-	}
-
-	err = cfgOps.SaveCfgToFile()
-	if err != nil {
-		return fmt.Errorf(`%s%s.%v`, getMsg(language, source, "write_data_fail"), configOperatorKey, err)
-	}
-	return nil
-}
-
-func AddSourceConfKeyField(plgName, confKey, language string, content []byte) error {
-	configOperatorKey := fmt.Sprintf(SourceCfgOperatorKeyTemplate, plgName)
-	return addConfKeyField(configOperatorKey, confKey, language, content)
-}
-
-func AddConnectionConfKeyField(plgName, confKey, language string, content []byte) error {
-	configOperatorKey := fmt.Sprintf(ConnectionCfgOperatorKeyTemplate, plgName)
-	return addConfKeyField(configOperatorKey, confKey, language, content)
-}
-
-func addConfKeyField(configOperatorKey, confKey, language string, content []byte) error {
-
-	ConfigManager.lock.Lock()
-	defer ConfigManager.lock.Unlock()
-
-	reqField := make(map[string]interface{})
-	err := json.Unmarshal(content, &reqField)
-	if nil != err {
 		return err
 	}
 
-	cfgOps, found := ConfigManager.cfgOperators[configOperatorKey]
-	if !found {
-		return fmt.Errorf(`%s%s`, getMsg(language, source, "not_found_plugin"), configOperatorKey)
-	}
-
-	err = cfgOps.AddConfKeyField(confKey, reqField)
-	if err != nil {
-		return fmt.Errorf(`%s%s.%v`, getMsg(language, source, "not_found_confkey"), confKey, err)
-	}
-	err = cfgOps.SaveCfgToFile()
-	if err != nil {
-		return fmt.Errorf(`%s%s.%v`, getMsg(language, source, "write_data_fail"), configOperatorKey, err)
-	}
-	return nil
-}
-
-func DelSourceConfKeyField(plgName, confKey, language string, content []byte) error {
-	configOperatorKey := fmt.Sprintf(SourceCfgOperatorKeyTemplate, plgName)
-	return delConfKeyField(configOperatorKey, confKey, language, content)
-}
-
-func DelConnectionConfKeyField(plgName, confKey, language string, content []byte) error {
-	configOperatorKey := fmt.Sprintf(ConnectionCfgOperatorKeyTemplate, plgName)
-	return delConfKeyField(configOperatorKey, confKey, language, content)
-}
-
-func delConfKeyField(configOperatorKey, confKey, language string, content []byte) error {
-
-	ConfigManager.lock.Lock()
-	defer ConfigManager.lock.Unlock()
-
-	cfgOps, found := ConfigManager.cfgOperators[configOperatorKey]
-	if !found {
-		return fmt.Errorf(`%s%s`, getMsg(language, source, "not_found_plugin"), configOperatorKey)
-	}
-
-	reqField := make(map[string]interface{})
-	err := json.Unmarshal(content, &reqField)
-	if nil != err {
-		return fmt.Errorf(`%s%s.%v`, getMsg(language, source, "json_marshal_fail"), configOperatorKey, err)
-	}
-
-	err = cfgOps.DeleteConfKeyField(confKey, reqField)
-	if err != nil {
-		return fmt.Errorf(`%s%s.%v`, getMsg(language, source, "type_conversion_fail"), configOperatorKey, err)
-	}
 	err = cfgOps.SaveCfgToFile()
 	if err != nil {
 		return fmt.Errorf(`%s%s.%v`, getMsg(language, source, "write_data_fail"), configOperatorKey, err)
 	}
 	return nil
-
 }
diff --git a/internal/meta/yaml_config_ops.go b/internal/meta/yaml_config_ops.go
index 5a50166c..9688db74 100644
--- a/internal/meta/yaml_config_ops.go
+++ b/internal/meta/yaml_config_ops.go
@@ -159,12 +159,7 @@ func (c *ConfigKeys) AddConfKey(confKey string, reqField map[string]interface{})
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
-	_, ok := c.cf[confKey]
-	if ok {
-		return fmt.Errorf("confKey %s already exist", confKey)
-	} else {
-		c.cf[confKey] = reqField
-	}
+	c.cf[confKey] = reqField
 
 	return nil
 }
diff --git a/internal/plugin/native/manager.go b/internal/plugin/native/manager.go
index 05faa446..b661deda 100644
--- a/internal/plugin/native/manager.go
+++ b/internal/plugin/native/manager.go
@@ -231,7 +231,18 @@ func (rr *Manager) Register(t plugin2.PluginType, j plugin2.Plugin) error {
 			return fmt.Errorf("invalid name %s: duplicate", name)
 		}
 	}
+
 	var err error
+	zipPath := path.Join(rr.pluginDir, name+".zip")
+	var unzipFiles []string
+	//clean up: delete zip file and unzip files in error
+	defer os.Remove(zipPath)
+	//download
+	err = httpx.DownloadFile(zipPath, uri)
+	if err != nil {
+		return fmt.Errorf("fail to download file %s: %s", uri, err)
+	}
+
 	if t == plugin2.FUNCTION {
 		if len(j.GetSymbols()) > 0 {
 			err = rr.db.Set(name, j.GetSymbols())
@@ -247,15 +258,6 @@ func (rr *Manager) Register(t plugin2.PluginType, j plugin2.Plugin) error {
 		return err
 	}
 
-	zipPath := path.Join(rr.pluginDir, name+".zip")
-	var unzipFiles []string
-	//clean up: delete zip file and unzip files in error
-	defer os.Remove(zipPath)
-	//download
-	err = httpx.DownloadFile(zipPath, uri)
-	if err != nil {
-		return fmt.Errorf("fail to download file %s: %s", uri, err)
-	}
 	//unzip and copy to destination
 	unzipFiles, version, err := rr.install(t, name, zipPath, shellParas)
 	if err == nil && len(j.GetSymbols()) > 0 {
diff --git a/internal/server/rest.go b/internal/server/rest.go
index 786bbac5..ea8f9de7 100644
--- a/internal/server/rest.go
+++ b/internal/server/rest.go
@@ -15,7 +15,6 @@
 package server
 
 import (
-	"crypto/tls"
 	"encoding/json"
 	"fmt"
 	"github.com/gorilla/handlers"
@@ -30,7 +29,6 @@ import (
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/ast"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
-	"golang.org/x/net/html"
 	"io"
 	"io/ioutil"
 	"net/http"
@@ -130,14 +128,12 @@ func createRestServer(ip string, port int, needToken bool) *http.Server {
 	r.HandleFunc("/metadata/sources", sourcesMetaHandler).Methods(http.MethodGet)
 	r.HandleFunc("/metadata/sources/{name}", sourceMetaHandler).Methods(http.MethodGet)
 	r.HandleFunc("/metadata/sources/yaml/{name}", sourceConfHandler).Methods(http.MethodGet)
-	r.HandleFunc("/metadata/sources/{name}/confKeys/{confKey}", sourceConfKeyHandler).Methods(http.MethodDelete, http.MethodPost)
-	r.HandleFunc("/metadata/sources/{name}/confKeys/{confKey}/field", sourceConfKeyFieldsHandler).Methods(http.MethodDelete, http.MethodPost)
+	r.HandleFunc("/metadata/sources/{name}/confKeys/{confKey}", sourceConfKeyHandler).Methods(http.MethodDelete, http.MethodPut)
 
 	r.HandleFunc("/metadata/connections", connectionsMetaHandler).Methods(http.MethodGet)
 	r.HandleFunc("/metadata/connections/{name}", connectionMetaHandler).Methods(http.MethodGet)
 	r.HandleFunc("/metadata/connections/yaml/{name}", connectionConfHandler).Methods(http.MethodGet)
-	r.HandleFunc("/metadata/connections/{name}/confKeys/{confKey}", connectionConfKeyHandler).Methods(http.MethodDelete, http.MethodPost)
-	r.HandleFunc("/metadata/connections/{name}/confKeys/{confKey}/field", connectionConfKeyFieldsHandler).Methods(http.MethodDelete, http.MethodPost)
+	r.HandleFunc("/metadata/connections/{name}/confKeys/{confKey}", connectionConfKeyHandler).Methods(http.MethodDelete, http.MethodPut)
 
 	r.HandleFunc("/services", servicesHandler).Methods(http.MethodGet, http.MethodPost)
 	r.HandleFunc("/services/functions", serviceFunctionsHandler).Methods(http.MethodGet)
@@ -659,13 +655,8 @@ func prebuildPluginsHandler(w http.ResponseWriter, r *http.Request, t plugin.Plu
 		os := "debian"
 		if strings.Contains(prettyName, "DEBIAN") {
 			hosts := conf.Config.Basic.PluginHosts
-			ptype := "sources"
-			if t == plugin.SINK {
-				ptype = "sinks"
-			} else if t == plugin.FUNCTION {
-				ptype = "functions"
-			}
-			if err, plugins := fetchPluginList(hosts, ptype, os, runtime.GOARCH); err != nil {
+
+			if err, plugins := fetchPluginList(t, hosts, os, runtime.GOARCH); err != nil {
 				handleError(w, err, "", logger)
 			} else {
 				jsonResponse(plugins, w, logger)
@@ -679,10 +670,24 @@ func prebuildPluginsHandler(w http.ResponseWriter, r *http.Request, t plugin.Plu
 	}
 }
 
-func fetchPluginList(hosts, ptype, os, arch string) (err error, result map[string]string) {
+var NativeSourcePlugin = []string{"random", "zmq"}
+var NativeSinkPlugin = []string{"file", "image", "influx", "redis", "tdengine", "zmq"}
+var NativeFunctionPlugin = []string{"accumulateWordCount", "countPlusOne", "echo", "geohash", "image", "labelImage"}
+
+func fetchPluginList(t plugin.PluginType, hosts, os, arch string) (err error, result map[string]string) {
+	ptype := "sources"
+	plugins := NativeSourcePlugin
+	if t == plugin.SINK {
+		ptype = "sinks"
+		plugins = NativeSinkPlugin
+	} else if t == plugin.FUNCTION {
+		ptype = "functions"
+		plugins = NativeFunctionPlugin
+	}
+
 	if hosts == "" || ptype == "" || os == "" {
 		logger.Errorf("Invalid parameter value: hosts %s, ptype %s or os: %s should not be empty.", hosts, ptype, os)
-		return fmt.Errorf("Invalid configruation for plugin host in kuiper.yaml."), nil
+		return fmt.Errorf("invalid configruation for plugin host in kuiper.yaml"), nil
 	}
 	result = make(map[string]string)
 	hostsArr := strings.Split(hosts, ",")
@@ -691,73 +696,14 @@ func fetchPluginList(hosts, ptype, os, arch string) (err error, result map[strin
 		tmp := []string{host, "kuiper-plugins", version, os, ptype}
 		//The url is similar to http://host:port/kuiper-plugins/0.9.1/debian/sinks/
 		url := strings.Join(tmp, "/")
-		timeout := time.Duration(30 * time.Second)
-		client := &http.Client{
-			Timeout: timeout,
-			Transport: &http.Transport{
-				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-			},
-		}
-		resp, err := client.Get(url)
-		logger.Infof("Trying to fetch plugins from url: %s\n", url)
-
-		if err != nil {
-			return err, nil
-		}
-		defer resp.Body.Close()
 
-		if resp.StatusCode != http.StatusOK {
-			return fmt.Errorf("Cannot fetch plugin list from %s, with status error: %v", url, resp.StatusCode), nil
-		}
-		data, err := ioutil.ReadAll(resp.Body)
-		if err != nil {
-			return err, nil
-		}
-		plugins := extractFromHtml(string(data), arch)
 		for _, p := range plugins {
-			//If already existed, using the existed.
-			if _, ok := result[p]; !ok {
-				result[p] = url + "/" + p + "_" + arch + ".zip"
-			}
-			logger.Debugf("Plugin %s, download address is %s\n", p, result[p])
+			result[p] = url + "/" + p + "_" + arch + ".zip"
 		}
 	}
 	return
 }
 
-func extractFromHtml(content, arch string) []string {
-	plugins := []string{}
-	htmlTokens := html.NewTokenizer(strings.NewReader(content))
-loop:
-	for {
-		tt := htmlTokens.Next()
-		switch tt {
-		case html.ErrorToken:
-			break loop
-		case html.StartTagToken:
-			t := htmlTokens.Token()
-			isAnchor := t.Data == "a"
-			if isAnchor {
-				found := false
-				for _, prop := range t.Attr {
-					if strings.ToUpper(prop.Key) == "HREF" {
-						if strings.HasSuffix(prop.Val, "_"+arch+".zip") {
-							if index := strings.LastIndex(prop.Val, "_"); index != -1 {
-								plugins = append(plugins, prop.Val[0:index])
-							}
-						}
-						found = true
-					}
-				}
-				if !found {
-					logger.Infof("Invalid plugin download link %s", t)
-				}
-			}
-		}
-	}
-	return plugins
-}
-
 //list sink plugin
 func sinksMetaHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
@@ -887,7 +833,7 @@ func sourceConfKeyHandler(w http.ResponseWriter, r *http.Request) {
 	switch r.Method {
 	case http.MethodDelete:
 		err = meta.DelSourceConfKey(pluginName, confKey, language)
-	case http.MethodPost:
+	case http.MethodPut:
 		v, err1 := ioutil.ReadAll(r.Body)
 		if err1 != nil {
 			handleError(w, err, "Invalid body", logger)
@@ -913,7 +859,7 @@ func connectionConfKeyHandler(w http.ResponseWriter, r *http.Request) {
 	switch r.Method {
 	case http.MethodDelete:
 		err = meta.DelConnectionConfKey(pluginName, confKey, language)
-	case http.MethodPost:
+	case http.MethodPut:
 		v, err1 := ioutil.ReadAll(r.Body)
 		if err1 != nil {
 			handleError(w, err1, "Invalid body", logger)
@@ -927,58 +873,6 @@ func connectionConfKeyHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-//Del and Update field of confkey
-func sourceConfKeyFieldsHandler(w http.ResponseWriter, r *http.Request) {
-	defer r.Body.Close()
-	var err error
-	vars := mux.Vars(r)
-	pluginName := vars["name"]
-	confKey := vars["confKey"]
-	v, err := ioutil.ReadAll(r.Body)
-	if err != nil {
-		handleError(w, err, "Invalid body", logger)
-		return
-	}
-
-	language := getLanguage(r)
-	switch r.Method {
-	case http.MethodDelete:
-		err = meta.DelSourceConfKeyField(pluginName, confKey, language, v)
-	case http.MethodPost:
-		err = meta.AddSourceConfKeyField(pluginName, confKey, language, v)
-	}
-	if err != nil {
-		handleError(w, err, "", logger)
-		return
-	}
-}
-
-//Del and Update field of confkey
-func connectionConfKeyFieldsHandler(w http.ResponseWriter, r *http.Request) {
-	defer r.Body.Close()
-	var err error
-	vars := mux.Vars(r)
-	pluginName := vars["name"]
-	confKey := vars["confKey"]
-	v, err := ioutil.ReadAll(r.Body)
-	if err != nil {
-		handleError(w, err, "Invalid body", logger)
-		return
-	}
-
-	language := getLanguage(r)
-	switch r.Method {
-	case http.MethodDelete:
-		err = meta.DelConnectionConfKeyField(pluginName, confKey, language, v)
-	case http.MethodPost:
-		err = meta.AddConnectionConfKeyField(pluginName, confKey, language, v)
-	}
-	if err != nil {
-		handleError(w, err, "", logger)
-		return
-	}
-}
-
 func getLanguage(r *http.Request) string {
 	language := r.Header.Get("Content-Language")
 	if 0 == len(language) {
diff --git a/internal/topo/topo.go b/internal/topo/topo.go
index ec5e04fe..f303307e 100644
--- a/internal/topo/topo.go
+++ b/internal/topo/topo.go
@@ -154,7 +154,7 @@ func (s *Topo) Open() <-chan error {
 		defer s.mu.Unlock()
 		var err error
 		if s.store, err = state.CreateStore(s.name, s.qos); err != nil {
-			fmt.Println(err)
+			s.ctx.GetLogger().Errorf("topo %s create store error %v", s.name, err)
 			s.drain <- err
 			return
 		}


Test patch:
diff --git a/internal/meta/yamlConfigMeta_test.go b/internal/meta/yamlConfigMeta_test.go
index 172b01cc..80fbeec8 100644
--- a/internal/meta/yamlConfigMeta_test.go
+++ b/internal/meta/yamlConfigMeta_test.go
@@ -54,35 +54,7 @@ func TestYamlConfigMeta_Ops(t *testing.T) {
 
 	//Exist ConfigKey , fail
 	err = AddSourceConfKey(plgName, "new", "en_US", []byte(addData))
-	if err == nil {
-		t.Error("should return error when overwrite exist config key")
-	}
-
-	addData1 := `{"interval":10000, "timeout":200}`
-
-	// no exist key, fail
-	noExistKey := "noexist"
-	err = AddSourceConfKeyField(noExistKey, "new", "en_US", []byte(addData1))
-	if err == nil {
-		t.Error("should return error when no exist key")
-	}
-
-	// exist key, success
-	ExistKey := plgName
-	err = AddSourceConfKeyField(ExistKey, "new", "en_US", []byte(addData1))
-	if err != nil {
-		t.Error(err)
-	}
-
-	// exist key, success
-	err = DelSourceConfKeyField(ExistKey, "new", "en_US", []byte(addData1))
 	if err != nil {
-		t.Error(err)
-	}
-
-	// exist key, success
-	err = DelSourceConfKey(ExistKey, "new", "en_US")
-	if err != nil {
-		t.Error(err)
+		t.Error("should overwrite exist config key")
 	}
 }
diff --git a/internal/server/rest_test.go b/internal/server/rest_test.go
index 37a4f4f2..642b7dff 100644
--- a/internal/server/rest_test.go
+++ b/internal/server/rest_test.go
@@ -16,240 +16,62 @@ package server
 
 import (
 	"fmt"
-	"net/http"
-	"net/http/httptest"
+	"github.com/lf-edge/ekuiper/internal/plugin"
 	"reflect"
-	"strings"
 	"testing"
 )
 
-func TestParseHtml(t1 *testing.T) {
-	var tests = []struct {
-		html    string
-		plugins []string
-		arch    string
-		error   string
+func Test_fetchPluginList(t *testing.T) {
+	version = "1.4.0"
+	type args struct {
+		t     plugin.PluginType
+		hosts string
+		os    string
+		arch  string
+	}
+	tests := []struct {
+		name    string
+		args    args
+		wantErr error
 	}{
 		{
-			html: `<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 3.2 Final//EN"><html>
-			<title>Directory listing for enterprise: /4.1.1/</title>
-			<meta http-equiv="Content-Type" content="text/html; charset=utf-8">
-			<meta name="robots" content="noindex,nofollow">
-			<body>
-			<h2>Directory listing for enterprise: /4.1.1/</h2>
-			<hr>
-			<ul>
-				<li><a href="file_386.zip">file_386.zip</a>
-				<li><a href="file_amd64.zip">file_amd64.zip</a>
-				<li><a href="file_arm.zip">file_arm.zip</a>
-				<li><a href="file_arm64.zip">file_arm64.zip</a>
-				<li><a href="file_ppc64le.zip">file_ppc64le.zip</a>
-
-				<li><a href="influx_386.zip">influx_386.zip</a>
-				<li><a href="influx_amd64.zip">influx_amd64.zip</a>
-				<li><a href="influx_arm.zip">influx_arm.zip</a>
-				<li><a href="influx_arm64.zip">influx_arm64.zip</a>
-				<li><a href="influx_ppc64le.zip">influx_ppc64le.zip</a>
-			</ul>
-			<hr>
-			</body>
-			</html>
-			`,
-			arch:    "arm64",
-			plugins: []string{"file", "influx"},
-			error:   "",
-		},
-
-		{
-			html: `<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 3.2 Final//EN"><html>
-			<title>Directory listing for enterprise: /4.1.1/</title>
-			<meta http-equiv="Content-Type" content="text/html; charset=utf-8">
-			<meta name="robots" content="noindex,nofollow">
-			<body>
-			<h2>Directory listing for enterprise: /4.1.1/</h2>
-			<hr>
-			<ul>
-				<li><a href="file_386.zip">file_386.zip</a>
-				<li><a href="file_amd64.zip">file_amd64.zip</a>
-				<li><a href="file_arm.zip">file_arm.zip</a>
-				<li><a href="file_arm64.zip">file_arm64.zip</a>
-				<li><a href="file_ppc64le.zip">file_ppc64le.zip</a>
-
-				<li><a href="influx_386.zip">influx_386.zip</a>
-				<li><a href="influx_amd64.zip">influx_amd64.zip</a>
-				<li><a href="influx_arm.zip">influx_arm.zip</a>
-				<li><a href="influx_arm64.zip">influx_arm64.zip</a>
-				<li><a href="influx_ppc64le.zip">influx_ppc64le.zip</a>
-			</ul>
-			<hr>
-			</body>
-			</html>
-			`,
-			arch:    "arm7",
-			plugins: []string{},
-			error:   "",
-		},
-
-		{
-			html: `<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 3.2 Final//EN"><html>
-			<title>Directory listing for enterprise: /4.1.1/</title>
-			<meta http-equiv="Content-Type" content="text/html; charset=utf-8">
-			<meta name="robots" content="noindex,nofollow">
-			<body>
-			<h2>Directory listing for enterprise: /4.1.1/</h2>
-			<hr>
-			<ul>
-				<li><a href="file_386.zip">file_386.zip</a>
-				<li><a href="file_amd64.zip">file_amd64.zip</a>
-				<li><a href="file_arm.zip">file_arm.zip</a>
-				<li><a href="file_arm64.zip">file_arm64.zip</a>
-				<li><a href="file_ppc64le.zip">file_ppc64le.zip</a>
-
-				<li><a href="influx_arm.zip">influx_arm.zip</a>
-				<li><a href="influx_arm64.zip">influx_arm64.zip</a>
-				<li><a href="influx_ppc64le.zip">influx_ppc64le.zip</a>
-			</ul>
-			<hr>
-			</body>
-			</html>
-			`,
-			arch:    "amd64",
-			plugins: []string{"file"},
-			error:   "",
+			"source",
+			args{
+				t:     plugin.SOURCE,
+				hosts: "http://127.0.0.1:8080",
+				os:    "debian",
+				arch:  "amd64",
+			},
+			nil,
 		},
-
 		{
-			html: `<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 3.2 Final//EN"><html>
-			<title>Directory listing for enterprise: /4.1.1/</title>
-			<meta http-equiv="Content-Type" content="text/html; charset=utf-8">
-			<meta name="robots" content="noindex,nofollow">
-			<body>
-			<h2>Directory listing for enterprise: /4.1.1/</h2>
-			<hr>
-			<ul>
-				
-			</ul>
-			<hr>
-			</body>
-			</html>
-			`,
-			arch:    "amd64",
-			plugins: []string{},
-			error:   "",
+			"sink",
+			args{
+				t:     plugin.SINK,
+				hosts: "http://127.0.0.1:8080",
+				os:    "debian",
+				arch:  "amd64",
+			},
+			nil,
 		},
-
 		{
-			html:    ``,
-			arch:    "amd64",
-			plugins: []string{},
-			error:   "",
+			"function",
+			args{
+				t:     plugin.FUNCTION,
+				hosts: "http://127.0.0.1:8080",
+				os:    "debian",
+				arch:  "amd64",
+			},
+			nil,
 		},
 	}
-
-	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
-	for i, t := range tests {
-		result := extractFromHtml(t.html, t.arch)
-		if t.error == "" && !reflect.DeepEqual(t.plugins, result) {
-			t1.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, t.html, t.plugins, result)
-		}
-	}
-}
-
-func TestFetchPluginList(t1 *testing.T) {
-	version = "0.9.1"
-	// Start a local HTTP server
-	server1 := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
-		// Send response to be tested
-		if _, err := rw.Write([]byte(`<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 3.2 Final//EN"><html>
-			<title>Directory listing for enterprise: /4.1.1/</title>
-			<meta http-equiv="Content-Type" content="text/html; charset=utf-8">
-			<meta name="robots" content="noindex,nofollow">
-			<body>
-			<h2>Directory listing for enterprise: /4.1.1/</h2>
-			<hr>
-			<ul>
-				<li><a href="file_386.zip">file_386.zip</a>
-				<li><a href="file_amd64.zip">file_amd64.zip</a>
-				<li><a href="file_arm.zip">file_arm.zip</a>
-				<li><a href="file_arm64.zip">file_arm64.zip</a>
-				<li><a href="file_ppc64le.zip">file_ppc64le.zip</a>
-
-				<li><a href="influx_386.zip">influx_386.zip</a>
-				<li><a href="influx_amd64.zip">influx_amd64.zip</a>
-				<li><a href="influx_arm.zip">influx_arm.zip</a>
-				<li><a href="influx_arm64.zip">influx_arm64.zip</a>
-				<li><a href="influx_ppc64le.zip">influx_ppc64le.zip</a>
-			</ul>
-			<hr>
-			</body>
-			</html>
-			`)); err != nil {
-			fmt.Printf("%s", err)
-		}
-
-	}))
-
-	server2 := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
-		// Send response to be tested
-		if _, err := rw.Write([]byte(`<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 3.2 Final//EN"><html>
-			<title>Directory listing for enterprise: /4.1.1/</title>
-			<meta http-equiv="Content-Type" content="text/html; charset=utf-8">
-			<meta name="robots" content="noindex,nofollow">
-			<body>
-			<h2>Directory listing for enterprise: /4.1.1/</h2>
-			<hr>
-			<ul>
-				<li><a href="file_arm64.zip">file_arm64.zip</a>
-
-				<li><a href="zmq_386.zip">influx_386.zip</a>
-				<li><a href="zmq_amd64.zip">influx_amd64.zip</a>
-				<li><a href="zmq_arm.zip">influx_arm.zip</a>
-				<li><a href="zmq_arm64.zip">influx_arm64.zip</a>
-				<li><a href="zmq_ppc64le.zip">influx_ppc64le.zip</a>
-			</ul>
-			<hr>
-			</body>
-			</html>
-			`)); err != nil {
-			fmt.Printf("%s", err)
-		}
-
-	}))
-
-	// Close the server when test finishes
-	defer server2.Close()
-
-	if e, r := fetchPluginList(strings.Join([]string{server1.URL, server2.URL}, ","), "sinks", "alpine", "arm64"); e != nil {
-		t1.Errorf("Error: %v", e)
-	} else {
-		exp := map[string]string{
-			"file":   server1.URL + "/kuiper-plugins/" + version + "/alpine/sinks/file_arm64.zip",
-			"influx": server1.URL + "/kuiper-plugins/" + version + "/alpine/sinks/influx_arm64.zip",
-			"zmq":    server2.URL + "/kuiper-plugins/" + version + "/alpine/sinks/zmq_arm64.zip",
-		}
-		if !reflect.DeepEqual(exp, r) {
-			t1.Errorf("result mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", exp, r)
-		}
-	}
-
-	if e, r := fetchPluginList(strings.Join([]string{server2.URL}, ","), "sinks", "alpine", "arm64"); e != nil {
-		t1.Errorf("Error: %v", e)
-	} else {
-		exp := map[string]string{
-			"zmq":  server2.URL + "/kuiper-plugins/" + version + "/alpine/sinks/zmq_arm64.zip",
-			"file": server2.URL + "/kuiper-plugins/" + version + "/alpine/sinks/file_arm64.zip",
-		}
-		if !reflect.DeepEqual(exp, r) {
-			t1.Errorf("result mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", exp, r)
-		}
-	}
-
-	if e, r := fetchPluginList(strings.Join([]string{server1.URL, server2.URL}, ","), "sinks", "alpine", "armv7"); e != nil {
-		t1.Errorf("Error: %v", e)
-	} else {
-		exp := map[string]string{}
-		if !reflect.DeepEqual(exp, r) {
-			t1.Errorf("result mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", exp, r)
-		}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			gotErr, gotResult := fetchPluginList(tt.args.t, tt.args.hosts, tt.args.os, tt.args.arch)
+			if !reflect.DeepEqual(gotErr, tt.wantErr) {
+				t.Errorf("fetchPluginList() gotErr = %v, want %v", gotErr, tt.wantErr)
+			}
+			fmt.Printf("%v", gotResult)
+		})
 	}
 }