Merge branch 'dev/0.9.1' into func

common/os_util.go

@@ -0,0 +1,135 @@
+package common
+
+import (
+	"bufio"
+	"bytes"
+	"errors"
+	"os"
+	"strings"
+)
+
+const EtcOsRelease string = "/etc/os-release"
+const UsrLibOsRelease string = "/usr/lib/os-release"
+
+// Read and return os-release, trying EtcOsRelease, followed by UsrLibOsRelease.
+// err will contain an error message if neither file exists or failed to parse
+func Read() (osrelease map[string]string, err error) {
+	osrelease, err = ReadFile(EtcOsRelease)
+	if err != nil {
+		osrelease, err = ReadFile(UsrLibOsRelease)
+	}
+	return
+}
+
+// Similar to Read(), but takes the name of a file to load instead
+func ReadFile(filename string) (osrelease map[string]string, err error) {
+	osrelease = make(map[string]string)
+	err = nil
+
+	lines, err := parseFile(filename)
+	if err != nil {
+		return
+	}
+
+	for _, v := range lines {
+		key, value, err := parseLine(v)
+		if err == nil {
+			osrelease[key] = value
+		}
+	}
+	return
+}
+
+// ReadString is similar to Read(), but takes a string to load instead
+func ReadString(content string) (osrelease map[string]string, err error) {
+	osrelease = make(map[string]string)
+	err = nil
+
+	lines, err := parseString(content)
+	if err != nil {
+		return
+	}
+
+	for _, v := range lines {
+		key, value, err := parseLine(v)
+		if err == nil {
+			osrelease[key] = value
+		}
+	}
+	return
+}
+
+func parseFile(filename string) (lines []string, err error) {
+	file, err := os.Open(filename)
+	if err != nil {
+		return nil, err
+	}
+	defer file.Close()
+
+	scanner := bufio.NewScanner(file)
+	for scanner.Scan() {
+		lines = append(lines, scanner.Text())
+	}
+	return lines, scanner.Err()
+}
+
+func parseString(content string) (lines []string, err error) {
+	in := bytes.NewBufferString(content)
+	reader := bufio.NewReader(in)
+	scanner := bufio.NewScanner(reader)
+
+	for scanner.Scan() {
+		lines = append(lines, scanner.Text())
+	}
+	return lines, scanner.Err()
+
+}
+
+func parseLine(line string) (key string, value string, err error) {
+	err = nil
+
+	// skip empty lines
+	if len(line) == 0 {
+		err = errors.New("Skipping: zero-length")
+		return
+	}
+
+	// skip comments
+	if line[0] == '#' {
+		err = errors.New("Skipping: comment")
+		return
+	}
+
+	// try to split string at the first '='
+	splitString := strings.SplitN(line, "=", 2)
+	if len(splitString) != 2 {
+		err = errors.New("Can not extract key=value")
+		return
+	}
+
+	// trim white space from key and value
+	key = splitString[0]
+	key = strings.Trim(key, " ")
+	value = splitString[1]
+	value = strings.Trim(value, " ")
+
+	// Handle double quotes
+	if strings.ContainsAny(value, `"`) {
+		first := string(value[0:1])
+		last := string(value[len(value)-1:])
+
+		if first == last && strings.ContainsAny(first, `"'`) {
+			value = strings.TrimPrefix(value, `'`)
+			value = strings.TrimPrefix(value, `"`)
+			value = strings.TrimSuffix(value, `'`)
+			value = strings.TrimSuffix(value, `"`)
+		}
+	}
+
+	// expand anything else that could be escaped
+	value = strings.Replace(value, `\"`, `"`, -1)
+	value = strings.Replace(value, `\$`, `$`, -1)
+	value = strings.Replace(value, `\\`, `\`, -1)
+	value = strings.Replace(value, "\\`", "`", -1)
+	return
+}

common/util.go

@@ -67,6 +67,7 @@ type KuiperConf struct {
 		RestTls        *tlsConf `yaml:"restTls"`
 		Prometheus     bool     `yaml:"prometheus"`
 		PrometheusPort int      `yaml:"prometheusPort"`
+		PluginHosts    string   `yaml:pluginHosts`
 	}
 	Rule api.RuleOption
 	Sink struct {

docs/en_US/operation/configuration_file.md

@@ -47,7 +47,58 @@ basic:
   prometheus: true
   prometheusPort: 20499
 ```
-For such a default configuration, Kuiper will export metrics and serve prometheus at ``http://localhost:20499/metrics``
+For such a default configuration, Kuiper will export metrics and serve prometheus at `http://localhost:20499/metrics`
+
+## Pluginhosts
+
+The URL where hosts all of pre-build plugins. By default it's at `packages.emqx.io`. There could be several hosts (host can be separated with comma), if same package could be found in the several hosts, then the package in the 1st host will have the highest priority.
+
+```yaml
+pluginHosts: https://packages.emqx.io
+```
+
+It could be also as following, you can specify a local repository, and the plugin in that repository will have higher priorities.
+
+```yaml
+pluginHosts: https://local.repo.net, https://packages.emqx.io
+```
+
+The directory structure of the plugins should be similar as following.
+
+```
+http://host:port/kuiper-plugins/0.9.1/sinks/alpine
+```
+
+The content of the page should be similar as below.
+
+```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>
+```
+
+
 
 ## Sink configurations
 

docs/en_US/restapi/plugins.md

@@ -127,4 +127,24 @@ DELETE http://localhost:8080/plugins/functions/{name}
 The user can pass a query parameter to decide if Kuiper should be stopped after a delete in order to make the deletion take effect. The parameter is `restart` and only when the value is `1` will the Kuiper be stopped. The user has to manually restart it.
 ```shell
 DELETE http://localhost:8080/plugins/sources/{name}?restart=1
+```
+
+## Get the available plugins
+
+According to the configuration `pluginHosts` in file `etc/kuiper.yaml` ,  it returns the plugins list that can be installed at local run Kuiper instance. By default, it get the list from `https://packages.emqx.io` .
+
+```
+GET http://localhost:9081/plugins/sources/prebuild
+GET http://localhost:9081/plugins/sinks/prebuild
+GET http://localhost:9081/plugins/functions/prebuild
+```
+
+The sample result is as following, and the key is plugin name, the value is plugin download address.
+
+```json
+{
+  "file": "http://127.0.0.1:63767/kuiper-plugins/0.9.1/sinks/alpine/file_arm64.zip",
+  "influx": "http://127.0.0.1:63767/kuiper-plugins/0.9.1/sinks/alpine/influx_arm64.zip",
+  "zmq": "http://127.0.0.1:63768/kuiper-plugins/0.9.1/sinks/alpine/zmq_arm64.zip"
+}
 ```

docs/zh_CN/operation/configuration_file.md

@@ -48,7 +48,58 @@ basic:
 ```
 在如上默认配置中，Kuiper 暴露于 Prometheusd 运行指标可通过 `http://localhost:20499/metrics` 访问。
 
-## Sink configurations
+## Pluginhosts 配置
+
+The URL where hosts all of pre-build plugins. By default it's at `packages.emqx.io`. There could be several hosts (host can be separated with comma), if same package could be found in the several hosts, then the package in the 1st host will have the highest priority.
+
+```yaml
+pluginHosts: https://packages.emqx.io
+```
+
+It could be also as following, you can specify a local repository, and the plugin in that repository will have higher priorities.
+
+```yaml
+pluginHosts: https://local.repo.net, https://packages.emqx.io
+```
+
+The directory structure of the plugins should be similar as following.
+
+```
+http://host:port/kuiper-plugins/0.9.1/sinks/alpine
+```
+
+The content of the page should be similar as below.
+
+```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>
+```
+
+
+
+## Sink 配置
 
 ```yaml
   #The cache persistence threshold size. If the message in sink cache is larger than 10, then it triggers persistence. If you find the remote system is slow to response, or sink throughput is small, then it's recommend to increase below 2 configurations.More memory is required with the increase of below 2 configurations.

docs/zh_CN/plugins/overview.md

@@ -32,7 +32,7 @@ source 的大部分属性用户通过对应的配置文件指定，用户无法
   这部分包含了插件的作者信息，插件开发者可以视情况提供这些信息，这部分信息会被展示在管理控制台的插件信息列表上。
 
       * name：名字
-      *  email：电子邮件地址
+      * email：电子邮件地址
       * company：公司名称
       * website：公司网站地址
 
@@ -60,7 +60,7 @@ source 的大部分属性用户通过对应的配置文件指定，用户无法
 - control：控件类型，控制在界面中显示的控件类型；**该字段必须提供；**
   - text：文本输入框
   - text-area：文字编辑区域
-  - list-box：列表框
+  - list：列表框
   - radio：单选框
 - helpUrl：如果有关于该属性更详细的帮助，可以在此指定；该字段可选；
   - en_US：英文文档帮助地址
@@ -71,6 +71,13 @@ source 的大部分属性用户通过对应的配置文件指定，用户无法
 - label：控件针对的标签控件；**该字段必须提供；**
   - en_US
   - zh_CN
+- type：字段类型；**该字段必须提供；**
+
+  * string：字符串
+  * float：小数
+  * int：整数
+  * list_object：列表，元素是结构体
+  * list_string：列表，元素是字符串
 - values：如果控件类型为 `list-box` 或者 `radio`，**该字段必须提供；**
   - 数组：数据类型可以为数字、字符、布尔等
 
@@ -202,8 +209,8 @@ source 的大部分属性用户通过对应的配置文件指定，用户无法
 - control：控件类型，控制在界面中显示的控件类型；**该字段必须提供；**
   - text：文本输入框
   - text-area：文字编辑区域
-  - list-box：列表框
-  - radio-button：单选框
+  - list：列表框
+  - radio：单选框
 - helpUrl：如果有关于该属性更详细的帮助，可以在此指定；该字段可选；
   - en_US：英文文档帮助地址
   - zh_CN：中文文档帮助地址
@@ -213,6 +220,14 @@ source 的大部分属性用户通过对应的配置文件指定，用户无法
 - label：控件针对的标签控件；**该字段必须提供；**
   - en_US
   - zh_CN
+- type：字段类型；**该字段必须提供；**
+  	* string：字符串
+  	* float：小数
+  	* int：整数
+  	* list_object：列表，元素是结构体
+  	* list_string：列表，元素是字符串
+  	* list_float：列表，元素是小数
+  	* list_int：列表，元素是整数
 - values：如果控件类型为 `list-box` 或者 `radio-button`，**该字段必须提供；**
   - 数组：数据类型可以为数字、字符、布尔等
 

docs/zh_CN/restapi/plugins.md

@@ -129,4 +129,24 @@ DELETE http://localhost:8080/plugins/functions/{name}
 
 ```shell
 DELETE http://localhost:8080/plugins/sources/{name}?restart=1
+```
+
+## 获取可安装的插件
+
+根据在 `etc/kuiper.yaml` 文件中 `pluginHosts` 的配置，获取适合本 Kuiper 实例运行的插件列表，缺省会从 `https://packages.emqx.io` 上去获取。
+
+```
+GET http://localhost:9081/plugins/sources/prebuild
+GET http://localhost:9081/plugins/sinks/prebuild
+GET http://localhost:9081/plugins/functions/prebuild
+```
+
+样例返回内容如下，其中键值为插件名称，值是插件的下载地址。
+
+```json
+{
+  "file": "http://127.0.0.1:63767/kuiper-plugins/0.9.1/sinks/alpine/file_arm64.zip",
+  "influx": "http://127.0.0.1:63767/kuiper-plugins/0.9.1/sinks/alpine/influx_arm64.zip",
+  "zmq": "http://127.0.0.1:63768/kuiper-plugins/0.9.1/sinks/alpine/zmq_arm64.zip"
+}
 ```

etc/kuiper.yaml

@@ -15,6 +15,10 @@ basic:
   # Prometheus settings
   prometheus: false
   prometheusPort: 20499
+  # The URL where hosts all of pre-build plugins. By default it's at packages.emqx.io
+  # There could be several hosts (host can be separated with comma), if same package could be found in the several hosts,
+  # then the package in the 1st host will have the highest priority.
+  pluginHosts: https://packages.emqx.io
 
 # The default options for all rules. Each rule can override this setting by defining its own option
 rule:

etc/mqtt_source.json

@@ -1,16 +1,16 @@
 {
 	"about": {
 		"trial": false,
-	"author": {
-		"name": "Jiyong Huang",
-		"email": "huangjy@emqx.io",
-		"company": "EMQ Technologies Co., Ltd",
-		"website": "https://www.emqx.io"
-	},
-	"helpUrl": {
-		"en_US": "https://github.com/emqx/kuiper/blob/master/docs/en_US/rules/sources/mqtt.md",
-		"zh_CN": "https://github.com/emqx/kuiper/blob/master/docs/zh_CN/rules/sources/mqtt.md"
-	},
+		"author": {
+			"name": "Jiyong Huang",
+			"email": "huangjy@emqx.io",
+			"company": "EMQ Technologies Co., Ltd",
+			"website": "https://www.emqx.io"
+		},
+		"helpUrl": {
+			"en_US": "https://github.com/emqx/kuiper/blob/master/docs/en_US/rules/sources/mqtt.md",
+			"zh_CN": "https://github.com/emqx/kuiper/blob/master/docs/zh_CN/rules/sources/mqtt.md"
+		},
 		"description": {
 			"en_US": "Kuiper provides built-in support for MQTT source stream, which can subscribe the message from MQTT broker and feed into the Kuiper processing pipeline.",
 			"zh_CN": "Kuiper 为 MQTT 源流提供了内置支持，流可以订阅来自 MQTT 代理的消息并输入Kuiper 处理管道。"
@@ -22,8 +22,9 @@
 			"name": "qos",
 			"default": 1,
 			"optional": false,
-			"control": "text",
+			"control": "select",
 			"type": "int",
+			"values": [0, 1, 2],
 			"hint": {
 				"en_US": "The default subscription QoS level.",
 				"zh_CN": "默认订阅 QoS 级别"
@@ -50,7 +51,7 @@
 			"name": "sharedSubscription",
 			"default": true,
 			"optional": true,
-			"control": "",
+			"control": "radio",
 			"type": "bool",
 			"hint": {
 				"en_US": "Whether use the shared subscription mode or not. If using the shared subscription mode, then there are multiple Kuiper process can be load balanced.",
@@ -64,8 +65,8 @@
 			"name": "servers",
 			"default": ["tcp://127.0.0.1:1883"],
 			"optional": true,
-			"control": "text",
-			"type": "string-list",
+			"control": "list",
+			"type": "list_string",
 			"hint": {
 				"en_US": "The server list for MQTT message broker. Currently, only ONE server can be specified.",
 				"zh_CN": "MQTT 消息代理的服务器列表。 当前，只能指定一个服务器。"

etc/sources/edgex.json

@@ -25,7 +25,7 @@
 				"values": [
 					"tcp"
 				],
-				"type": "list_string",
+				"type": "string",
 				"hint": {
 					"en_US": "The protocol. If it's not specified, then use default value 'tcp'.",
 					"zh_CN": "协议，如未指定，使用缺省值 tcp。"
@@ -161,8 +161,9 @@
 						"name": "Qos",
 						"default": "",
 						"optional": true,
-						"control": "text",
-						"type": "string",
+						"control": "select",
+						"type": "int",
+						"values": [0, 1, 2],
 						"hint": {
 							"en_US": "MQTT QoS",
 							"zh_CN": "MQTT 服务质量"
@@ -294,8 +295,8 @@
 					}
 				],
 				"optional": true,
-				"control": "text",
-				"type": "obj-list",
+				"control": "list",
+				"type": "list_object",
 				"hint": {
 					"en_US": "If MQTT message bus is used, some other optional configurations can be specified. Please notice that all of values in optional are string type, so values for these configurations should be string - such as KeepAlive: \"5000\".",
 					"zh_CN": "如果使用了 MQTT 消息总线，还可以指定别的一些可选配置项。请注意，所有在可选的配置项里指定的值都必须为**字符类型**，因此这里出现的所有的配置应该是字符类型的 - 例如 KeepAlive: \"5000\"。"

etc/sources/httppull.json

@@ -2,16 +2,16 @@
 	"libs": [],
 	"about": {
 		"trial": false,
-	"author": {
-		"name": "Jiyong Huang",
-		"email": "huangjy@emqx.io",
-		"company": "EMQ Technologies Co., Ltd",
-		"website": "https://www.emqx.io"
-	},
-	"helpUrl": {
-		"en_US": "https://github.com/emqx/kuiper/blob/master/docs/en_US/rules/sources/http_pull.md",
-		"zh_CN": "https://github.com/emqx/kuiper/blob/master/docs/zh_CN/rules/sources/http_pull.md"
-	},
+		"author": {
+			"name": "Jiyong Huang",
+			"email": "huangjy@emqx.io",
+			"company": "EMQ Technologies Co., Ltd",
+			"website": "https://www.emqx.io"
+		},
+		"helpUrl": {
+			"en_US": "https://github.com/emqx/kuiper/blob/master/docs/en_US/rules/sources/http_pull.md",
+			"zh_CN": "https://github.com/emqx/kuiper/blob/master/docs/zh_CN/rules/sources/http_pull.md"
+		},
 		"description": {
 			"en_US": "Kuiper provides built-in support for pulling HTTP source stream, which can pull the message from HTTP server broker and feed into the Kuiper processing pipeline.",
 			"zh_CN": "Kuiper 为提取 HTTP 源流提供了内置支持，该支持可从 HTTP 服务器代理提取消息并输入 Kuiper 处理管道。"
@@ -78,7 +78,7 @@
 			"name": "incremental",
 			"default": false,
 			"optional": false,
-			"control": "text",
+			"control": "radio",
 			"type": "bool",
 			"hint": {
 				"en_US": "If it's set to true, then will compare with last result; If response of two requests are the same, then will skip sending out the result.",
@@ -92,7 +92,7 @@
 			"name": "body",
 			"default": "{}",
 			"optional": false,
-			"control": "text",
+			"control": "textarea",
 			"type": "string",
 			"hint": {
 				"en_US": "The body of request",
@@ -118,24 +118,24 @@
 			}
 		}, {
 			"name": "headers",
-			"default": [{                                                                                 
-      "name": "Accept",                                                                
-      "default": "application/json",                                                                 
-      "optional": false,                                                                 
-      "control": "text",                                                                 
-      "type": "string",                                                                  
-      "hint": {                                                                          
-        "en_US": "HTTP headers",      
-        "zh_CN": "HTTP标头"              
-      },                                                                                 
-      "label": {                                                                         
-        "en_US": "HTTP headers",
-        "zh_CN": "HTTP标头"
-      }                                                                                  
-    }],
+			"default": [{                                                 
+				"name": "Accept",
+				"default": "application/json",
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "HTTP headers",
+					"zh_CN": "HTTP标头"      
+				},                                                              
+				"label": {
+					"en_US": "HTTP headers",
+					"zh_CN": "HTTP标头"
+				}                                                     
+			}],
 			"optional": false,
-			"control": "text",
-			"type": "string",
+			"control": "list",
+			"type": "list_object",
 			"hint": {
 				"en_US": "The HTTP request headers that you want to send along with the HTTP request.",
 				"zh_CN": "需要与 HTTP 请求一起发送的 HTTP 请求标头。"

etc/sources/random.json

@@ -86,8 +86,8 @@
           }
         ],
         "optional": true,
-        "control": "text",
-        "type": "string",
+        "control": "list",
+        "type": "list_object",
         "hint": {
           "en_US": "The pattern to be generated by the source",
           "zh_CN": "源生成的样式"

go.mod

@@ -18,6 +18,7 @@ require (
 	github.com/prometheus/client_golang v1.2.1
 	github.com/sirupsen/logrus v1.4.2
 	github.com/urfave/cli v1.22.0
+	golang.org/x/net v0.0.0-20191209160850-c0dbc17a3553
 )
 
 go 1.13

plugins/manager.go

@@ -277,16 +277,10 @@ func (m *Manager) Register(t PluginType, j *Plugin) error {
 		}
 		return fmt.Errorf("fail to unzip file %s: %s", uri, err)
 	}
-	confDir, err := common.GetConfLoc()
-	if nil != err {
-		return err
-	}
 	if SINK == t {
-		readSinkMetaFile(path.Join(confDir, PluginTypes[t], name+`.json`))
+		readSinkMetaFile(path.Join(m.etcDir, PluginTypes[t], name+`.json`))
 	} else if SOURCE == t {
-		readSourceMetaFile(path.Join(confDir, PluginTypes[t], name+`.json`))
-	} else if FUNCTION == t {
-		readFuncMetaFile(path.Join(confDir, PluginTypes[t], name+`.json`))
+		readSourceMetaFile(path.Join(m.etcDir, PluginTypes[t], name+`.json`))
 	}
 	m.registry.Store(t, name, version)
 	return nil
@@ -305,9 +299,7 @@ func (m *Manager) Delete(t PluginType, name string, stop bool) error {
 	paths := []string{
 		path.Join(m.pluginDir, PluginTypes[t], soFile),
 	}
-	if confDir, err := common.GetConfLoc(); nil == err {
-		os.Remove(path.Join(confDir, PluginTypes[t], name+".json"))
-	}
+	os.Remove(path.Join(m.etcDir, PluginTypes[t], name+".json"))
 	if t == SOURCE {
 		paths = append(paths, path.Join(m.etcDir, PluginTypes[t], name+".yaml"))
 	}
@@ -392,13 +384,8 @@ func (m *Manager) install(t PluginType, name string, src string) ([]string, stri
 			}
 			filenames = append(filenames, yamlPath)
 		} else if fileName == name+".json" {
-			confDir, err := common.GetConfLoc()
-			if nil != err {
-				return filenames, "", err
-			}
-			err = unzipTo(file, path.Join(confDir, PluginTypes[t], fileName))
-			if err != nil {
-				return filenames, "", err
+			if err := unzipTo(file, path.Join(m.etcDir, PluginTypes[t], fileName)); nil != err {
+				common.Log.Errorf("Failed to decompress the metadata %s file", fileName)
 			}
 		} else if soPrefix.Match([]byte(fileName)) {
 			soPath := path.Join(m.pluginDir, PluginTypes[t], fileName)

plugins/sourceMeta.go

@@ -154,9 +154,9 @@ func DelSourceConfKey(pluginName, confKey string) error {
 	return property.saveCf(pluginName)
 }
 
-func AddSourceConfKey(pluginName, confKey, content string) error {
+func AddSourceConfKey(pluginName, confKey string, content []byte) error {
 	reqField := make(map[string]interface{})
-	err := json.Unmarshal([]byte(content), &reqField)
+	err := json.Unmarshal(content, &reqField)
 	if nil != err {
 		return err
 	}
@@ -179,9 +179,9 @@ func AddSourceConfKey(pluginName, confKey, content string) error {
 	return property.saveCf(pluginName)
 }
 
-func AddSourceConfKeyField(pluginName, confKey, content string) error {
+func AddSourceConfKeyField(pluginName, confKey string, content []byte) error {
 	reqField := make(map[string]interface{})
-	err := json.Unmarshal([]byte(content), &reqField)
+	err := json.Unmarshal(content, &reqField)
 	if nil != err {
 		return err
 	}
@@ -243,9 +243,9 @@ func recursionDelMap(cf, fields map[string]interface{}) error {
 	return nil
 }
 
-func DelSourceConfKeyField(pluginName, confKey, content string) error {
+func DelSourceConfKeyField(pluginName, confKey string, content []byte) error {
 	reqField := make(map[string]interface{})
-	err := json.Unmarshal([]byte(content), &reqField)
+	err := json.Unmarshal(content, &reqField)
 	if nil != err {
 		return err
 	}

plugins/sourceMeta_test.go

@@ -44,14 +44,14 @@ func TestGetSourceMeta(t *testing.T) {
 	addData := `{"url":"127.0.0.1","method":"post","headers":{"Accept":"json"}}`
 	delData := `{"method":"","headers":{"Accept":""}}`
 
-	if err := AddSourceConfKey(g_plugin, "new", addData); nil != err {
+	if err := AddSourceConfKey(g_plugin, "new", []byte(addData)); nil != err {
 		t.Error(err)
 	}
 	if err := isAddData(addData, cf[`new`]); nil != err {
 		t.Error(err)
 	}
 
-	if err := DelSourceConfKeyField(g_plugin, "new", delData); nil != err {
+	if err := DelSourceConfKeyField(g_plugin, "new", []byte(delData)); nil != err {
 		t.Error(err)
 	}
 	if err := isDelData(delData, cf[`new`]); nil != err {

xstream/server/server/rest.go

@@ -8,10 +8,13 @@ import (
 	"github.com/emqx/kuiper/xstream/api"
 	"github.com/gorilla/handlers"
 	"github.com/gorilla/mux"
+	"golang.org/x/net/html"
 	"io"
 	"io/ioutil"
 	"net/http"
+	"os"
 	"runtime"
+	"strings"
 	"time"
 )
 
@@ -82,10 +85,14 @@ func createRestServer(port int) *http.Server {
 	r.HandleFunc("/rules/{name}/topo", getTopoRuleHandler).Methods(http.MethodGet)
 
 	r.HandleFunc("/plugins/sources", sourcesHandler).Methods(http.MethodGet, http.MethodPost)
+	r.HandleFunc("/plugins/sources/prebuild", prebuildSourcePlugins).Methods(http.MethodGet)
 	r.HandleFunc("/plugins/sources/{name}", sourceHandler).Methods(http.MethodDelete, http.MethodGet)
+
 	r.HandleFunc("/plugins/sinks", sinksHandler).Methods(http.MethodGet, http.MethodPost)
+	r.HandleFunc("/plugins/sinks/prebuild", prebuildSinkPlugins).Methods(http.MethodGet)
 	r.HandleFunc("/plugins/sinks/{name}", sinkHandler).Methods(http.MethodDelete, http.MethodGet)
 	r.HandleFunc("/plugins/functions", functionsHandler).Methods(http.MethodGet, http.MethodPost)
+	r.HandleFunc("/plugins/functions/prebuild", prebuildFuncsPlugins).Methods(http.MethodGet)
 	r.HandleFunc("/plugins/functions/{name}", functionHandler).Methods(http.MethodDelete, http.MethodGet)
 
 	r.HandleFunc("/metadata/functions", functionsMetaHandler).Methods(http.MethodGet)
@@ -413,6 +420,130 @@ func functionHandler(w http.ResponseWriter, r *http.Request) {
 	pluginHandler(w, r, plugins.FUNCTION)
 }
 
+func prebuildSourcePlugins(w http.ResponseWriter, r *http.Request) {
+	prebuildPluginsHandler(w, r, plugins.SOURCE)
+}
+
+func prebuildSinkPlugins(w http.ResponseWriter, r *http.Request) {
+	prebuildPluginsHandler(w, r, plugins.SINK)
+}
+
+func prebuildFuncsPlugins(w http.ResponseWriter, r *http.Request) {
+	prebuildPluginsHandler(w, r, plugins.FUNCTION)
+}
+
+func isOffcialDockerImage() bool {
+	if strings.ToLower(os.Getenv("MAINTAINER")) != "emqx.io" {
+		return false
+	}
+	return true
+}
+
+func prebuildPluginsHandler(w http.ResponseWriter, r *http.Request, t plugins.PluginType) {
+	if runtime.GOOS != "linux" {
+		handleError(w, fmt.Errorf("Plugins can be only installed at Linux."), "", logger)
+		return
+	} else if !isOffcialDockerImage() {
+		handleError(w, fmt.Errorf("Plugins can be only installed at official released Docker images."), "", logger)
+		return
+	} else if runtime.GOOS == "linux" {
+		osrelease, err := common.Read()
+		if err != nil {
+			logger.Infof("")
+			return
+		}
+		prettyName := strings.ToUpper(osrelease["PRETTY_NAME"])
+		if strings.Contains(prettyName, "ALPINE") || strings.Contains(prettyName, "DEBIAN") {
+			hosts := common.Config.Basic.PluginHosts
+			ptype := "sources"
+			if t == plugins.SINK {
+				ptype = "sinks"
+			} else if t == plugins.FUNCTION {
+				ptype = "functions"
+			}
+			if err, plugins := fetchPluginList(hosts, ptype, strings.ToLower(prettyName), runtime.GOARCH); err != nil {
+				handleError(w, err, "", logger)
+			} else {
+				jsonResponse(plugins, w, logger)
+			}
+		} else {
+			handleError(w, fmt.Errorf("Only ALPINE & DEBIAN docker images are supported."), "", logger)
+			return
+		}
+	} else {
+		handleError(w, fmt.Errorf("Please use official Kuiper docker images to install the plugins."), "", logger)
+	}
+}
+
+func fetchPluginList(hosts, ptype, os, arch string) (err error, result map[string]string) {
+	if hosts == "" || ptype == "" || os == "" {
+		return fmt.Errorf("Invalid parameter value: hosts, ptype and os value should not be empty."), nil
+	}
+	result = make(map[string]string)
+	hostsArr := strings.Split(hosts, ",")
+	for _, host := range hostsArr {
+		host := strings.Trim(host, " ")
+		tmp := []string{host, "kuiper-plugins", version, ptype, os}
+		//The url is similar to http://host:port/kuiper-plugins/0.9.1/sinks/alpine
+		url := strings.Join(tmp, "/")
+		resp, err := http.Get(url)
+
+		if err != nil {
+			return err, nil
+		}
+		defer resp.Body.Close()
+
+		if resp.StatusCode != http.StatusOK {
+			return fmt.Errorf("Status error: %v", 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"
+			}
+		}
+	}
+	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()
@@ -503,6 +634,7 @@ func sourceConfKeysHandler(w http.ResponseWriter, r *http.Request) {
 
 //Add  del confkey
 func sourceConfKeyHandler(w http.ResponseWriter, r *http.Request) {
+
 	defer r.Body.Close()
 	var ret interface{}
 	var err error
@@ -513,12 +645,12 @@ func sourceConfKeyHandler(w http.ResponseWriter, r *http.Request) {
 	case http.MethodDelete:
 		err = plugins.DelSourceConfKey(pluginName, confKey)
 	case http.MethodPost:
-		v, err := decodeStatementDescriptor(r.Body)
+		v, err := ioutil.ReadAll(r.Body)
 		if err != nil {
 			handleError(w, err, "Invalid body", logger)
 			return
 		}
-		err = plugins.AddSourceConfKey(pluginName, confKey, v.Sql)
+		err = plugins.AddSourceConfKey(pluginName, confKey, v)
 	}
 	if err != nil {
 		handleError(w, err, "metadata error", logger)
@@ -538,16 +670,16 @@ func sourceConfKeyFieldsHandler(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	pluginName := vars["name"]
 	confKey := vars["confKey"]
-	v, err := decodeStatementDescriptor(r.Body)
+	v, err := ioutil.ReadAll(r.Body)
 	if err != nil {
 		handleError(w, err, "Invalid body", logger)
 		return
 	}
 	switch r.Method {
 	case http.MethodDelete:
-		err = plugins.DelSourceConfKeyField(pluginName, confKey, v.Sql)
+		err = plugins.DelSourceConfKeyField(pluginName, confKey, v)
 	case http.MethodPost:
-		err = plugins.AddSourceConfKeyField(pluginName, confKey, v.Sql)
+		err = plugins.AddSourceConfKeyField(pluginName, confKey, v)
 	}
 	if err != nil {
 		handleError(w, err, "metadata error", logger)

xstream/server/server/rest_test.go

@@ -0,0 +1,241 @@
+package server
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"reflect"
+	"strings"
+	"testing"
+)
+
+func TestParseHtml(t1 *testing.T) {
+	var tests = []struct {
+		html    string
+		plugins []string
+		arch    string
+		error   string
+	}{
+		{
+			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:   "",
+		},
+
+		{
+			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:   "",
+		},
+
+		{
+			html:    ``,
+			arch:    "amd64",
+			plugins: []string{},
+			error:   "",
+		},
+	}
+
+	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 + "/sinks/alpine/file_arm64.zip",
+			"influx": server1.URL + "/kuiper-plugins/" + version + "/sinks/alpine/influx_arm64.zip",
+			"zmq":    server2.URL + "/kuiper-plugins/" + version + "/sinks/alpine/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 + "/sinks/alpine/zmq_arm64.zip",
+			"file": server2.URL + "/kuiper-plugins/" + version + "/sinks/alpine/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)
+		}
+	}
+}