Repository: lf-edge/ekuiper
Pull request #1396 (2022-09-07 02:37:50)

feat(plugin): add support for the sink plug-in of the influx V2 version
Signed-off-by: crui <crui@relaper.com>

elpsyr: /assign @ngjaying
ngjaying: @elpsyr @superrxan Is it better to name it influx2 or influxV2? Go convention is to MixedCaps as variable instead of underscore, [see this](https://go.dev/doc/effective_go#mixed-caps)
superrxan: to build this as an official release plugin 
should add build target in Makefile ``sinks/influx_v2`` where ``sinks`` is sinks folder and ``influx_v2`` is influx_v2 source code folder
if this section
``` shell
PLUGINS := sinks/file \
	sinks/influx \
	sinks/zmq \
	sinks/image \
	sinks/redis \
	sinks/sql   \
	sources/random \
	sources/zmq \
	sources/sql \
	sinks/tdengine \
	functions/accumulateWordCount \
	functions/countPlusOne \
	functions/image \
	functions/geohash \
	functions/echo \
	functions/labelImage
```

Patch:
diff --git a/docs/directory.json b/docs/directory.json
index f2045109..38a12d8b 100644
--- a/docs/directory.json
+++ b/docs/directory.json
@@ -288,6 +288,10 @@
 									"title": "InfluxDB 动作",
 									"path": "rules/sinks/plugin/influx"
 								},
+								{
+									"title": "InfluxDBV2 动作",
+									"path": "rules/sinks/plugin/influx2"
+								},
 								{
 									"title": "TDengine 动作",
 									"path": "rules/sinks/plugin/tdengine"
@@ -842,6 +846,10 @@
 									"title": "InfluxDB sink",
 									"path": "rules/sinks/plugin/influx"
 								},
+								{
+									"title": "InfluxDBV2 sink",
+									"path": "rules/sinks/plugin/influx2"
+								},
 								{
 									"title": "TDengine sink",
 									"path": "rules/sinks/plugin/tdengine"
diff --git a/docs/en_US/rules/sinks/overview.md b/docs/en_US/rules/sinks/overview.md
index b1bec2bf..c26645a1 100644
--- a/docs/en_US/rules/sinks/overview.md
+++ b/docs/en_US/rules/sinks/overview.md
@@ -24,7 +24,8 @@ The list of predefined sink plugins:
 
 - [Zero MQ sink](./plugin/zmq.md): sink to zero mq.
 - [File sink](./plugin/file.md): sink to a file.
-- [InfluxDB sink](./plugin/influx.md): sink to influx db.
+- [InfluxDB sink](./plugin/influx.md): sink to influx db `v1.x`. 
+- [InfluxDBV2 sink](./plugin/influx2.md): sink to influx db `v2.x`.
 - [Tdengine sink](./plugin/tdengine.md): sink to tdengine.
 - [Redis sink](./plugin/redis.md): sink to redis.
 - [Image sink](./plugin/image.md): sink to an image file. Only used to handle binary result.
diff --git a/docs/en_US/rules/sinks/plugin/influx2.md b/docs/en_US/rules/sinks/plugin/influx2.md
new file mode 100644
index 00000000..4a6b17a2
--- /dev/null
+++ b/docs/en_US/rules/sinks/plugin/influx2.md
@@ -0,0 +1,112 @@
+# InfluxDB Sink
+
+The sink will publish the result into a InfluxDB `V2.X` .
+
+## Compile & deploy plugin
+
+Please make following update before compile the plugin,
+
+- Add Influxdb library reference in `go.mod`.
+- Remove the first line `// +build plugins` of file `plugins/sinks/influx.go`.
+
+### build in shell
+```shell
+# cd $eKuiper_src
+# go build -trimpath -modfile extensions.mod --buildmode=plugin -o plugins/sinks/influx2.so extensions/sinks/influx/influx2.go
+# zip influx2.zip plugins/sinks/influx2.so
+# cp influx.zip /root/tomcat_path/webapps/ROOT/
+# bin/kuiper create plugin sink influx2 -f /tmp/influxPlugin.txt
+# bin/kuiper create rule influx2 -f /tmp/influxRule.txt
+```
+
+### build with image
+```
+docker build -t demo/plugins:v1 -f build/plugins/Dockerfile .
+docker run demo/plugins:v1
+docker cp  90eae15a7245:/workspace/_plugins/debian/sinks /tmp
+```
+Dockerfile like this：
+```
+## plase check go version that kuiper used
+ARG GO_VERSION=1.18.5
+FROM ghcr.io/lf-edge/ekuiper/base:$GO_VERSION-debian AS builder
+WORKDIR /workspace
+ADD . /workspace/
+RUN go env -w GOPROXY=https://goproxy.cn,direct
+RUN make plugins_c
+CMD ["sleep","3600"]
+```
+add this in Makefile：
+```
+PLUGINS_CUSTOM := sinks/influx2
+
+.PHONY: plugins_c $(PLUGINS_CUSTOM)
+plugins_c: $(PLUGINS_CUSTOM)
+
+$(PLUGINS_CUSTOM): PLUGIN_TYPE = $(word 1, $(subst /, , $@))
+$(PLUGINS_CUSTOM): PLUGIN_NAME = $(word 2, $(subst /, , $@))
+$(PLUGINS_CUSTOM):
+	@$(CURDIR)/build-plugins.sh $(PLUGIN_TYPE) $(PLUGIN_NAME)
+```
+
+Restart the eKuiper server to activate the plugin.
+
+## Properties
+
+| Property name | Optional | Description                                       |
+|---------------|----------|---------------------------------------------------|
+| addr          | true     | The addr of the InfluxDB                          |
+| measurement   | true     | The measurement of the InfluxDb (like table name) |
+| org           | false    | The InfluxDB organization                         |
+| bucket        | false    | The InfluxDB bucket                               |
+| token         | false    | The token of access InfluxDB                      |
+| tagKey        | true     | The tag key of the InfluxDB                       |
+| tagValue      | true     | The tag value of the InfluxDB                     |
+| fields        | true     | The column of the InfluxDB,split with ","         |
+## Sample usage
+
+Below is a sample for selecting temperature great than 50 degree, and some profiles only for your reference.
+
+### /tmp/influxRule.txt
+```json
+{
+  "id": "influx",
+  "sql": "SELECT * from  demo_stream where temperature > 50",
+  "actions": [
+    {
+      "log": {},
+      "influx2":{
+        "addr": "http://192.168.100.245:8086",
+        "token": "test_token",
+        "org": "admin",
+        "measurement": "test",
+        "bucket": "bucketName",
+        "tagKey": "tagKey",
+        "tagValue": "tagValue",
+        "fields": "humidity,temperature,pressure"
+      }
+    }
+  ]
+}
+```
+### /tmp/influxPlugin.txt
+```json
+{
+   "file":"http://localhost:8080/influx2.zip"
+ }
+```
+### plugins/go.mod
+```
+module plugins
+
+go 1.18
+
+require (
+        github.com/lf-edge/ekuiper v0.0.0-20220727015637-7d6f5c447110
+        github.com/influxdata/influxdb-client-go/v2 v2.10.0
+        github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839 // indirect
+)
+
+replace github.com/lf-edge/ekuiper => /root/goProject/kuiper
+
+```
diff --git a/docs/zh_CN/rules/sinks/overview.md b/docs/zh_CN/rules/sinks/overview.md
index 2a117273..b1d81104 100644
--- a/docs/zh_CN/rules/sinks/overview.md
+++ b/docs/zh_CN/rules/sinks/overview.md
@@ -24,7 +24,8 @@
 
 - [Zero MQ sink](./plugin/zmq.md)：输出到 Zero MQ 。
 - [File sink](./plugin/file.md)： 写入文件。
-- [InfluxDB sink](./plugin/influx.md)： 写入 Influx DB 。
+- [InfluxDB sink](./plugin/influx.md)： 写入 Influx DB `v1.x`。
+- [InfluxDBV2 sink](./plugin/influx2.md)： 写入 Influx DB `v2.x`。
 - [Tdengine sink](./plugin/tdengine.md)： 写入 Tdengine 。
 - [Redis sink](./plugin/redis.md)： 写入 redis 。
 - [Image sink](./plugin/image.md)： 写入一个图像文件。仅用于处理二进制结果。
diff --git a/docs/zh_CN/rules/sinks/plugin/influx2.md b/docs/zh_CN/rules/sinks/plugin/influx2.md
new file mode 100644
index 00000000..bc56a01b
--- /dev/null
+++ b/docs/zh_CN/rules/sinks/plugin/influx2.md
@@ -0,0 +1,110 @@
+# InfluxDB 目标（Sink）
+
+该插件将分析结果发送到 InfluxDB V2.X 中。
+## 编译插件&创建插件
+
+在编译之前，请对源代码做如下更改：
+
+- 在 `go.mod` 文件中增加对 InfluxDB 库文件的引用
+
+### 本地构建
+```shell
+# cd $eKuiper_src
+# go build -trimpath -modfile extensions.mod --buildmode=plugin -o plugins/sinks/influx2.so extensions/sinks/influx/influx2.go
+# zip influx2.zip plugins/sinks/influx2.so
+# cp influx.zip /root/tomcat_path/webapps/ROOT/
+# bin/kuiper create plugin sink influx2 -f /tmp/influxPlugin.txt
+# bin/kuiper create rule influx2 -f /tmp/influxRule.txt
+```
+
+### 镜像构建
+```
+docker build -t demo/plugins:v1 -f build/plugins/Dockerfile .
+docker run demo/plugins:v1
+docker cp  90eae15a7245:/workspace/_plugins/debian/sinks /tmp
+```
+Dockerfile 如下所示：
+```
+## plase check go version that kuiper used
+ARG GO_VERSION=1.18.5
+FROM ghcr.io/lf-edge/ekuiper/base:$GO_VERSION-debian AS builder
+WORKDIR /workspace
+ADD . /workspace/
+RUN go env -w GOPROXY=https://goproxy.cn,direct
+RUN make plugins_c
+CMD ["sleep","3600"]
+```
+在Makefile中添加：
+```
+PLUGINS_CUSTOM := sinks/influx2
+
+.PHONY: plugins_c $(PLUGINS_CUSTOM)
+plugins_c: $(PLUGINS_CUSTOM)
+
+$(PLUGINS_CUSTOM): PLUGIN_TYPE = $(word 1, $(subst /, , $@))
+$(PLUGINS_CUSTOM): PLUGIN_NAME = $(word 2, $(subst /, , $@))
+$(PLUGINS_CUSTOM):
+	@$(CURDIR)/build-plugins.sh $(PLUGIN_TYPE) $(PLUGIN_NAME)
+```
+
+重新启动 eKuiper 服务器以激活插件。
+
+## 属性
+
+| 属性名称        | 会否可选 | 说明                 |
+|-------------|------|--------------------|
+| addr        | 是    | InfluxDB的地址        |
+| measurement | 是    | InfluxDb的测量（如表名）   |
+| org         | 否    | InfluxDB存储组织       |
+| bucket      | 否    | InfluxDB存储bucket   |
+| token       | 否    | InfluxDB访问Token    |
+| tagKey      | 是    | InfluxDB的标签键       |
+| tagValue    | 是    | InfluxDB的标签值       |
+| fields      | 是    | InfluxDB的列名,用","隔开 |
+## 示例用法
+
+下面是选择温度大于50度的样本规则，和一些配置文件仅供参考。
+
+### ####/tmp/influxRule.txt
+```json
+{
+  "id": "influx",
+  "sql": "SELECT * from  demo_stream where temperature > 50",
+  "actions": [
+    {
+      "log": {},
+      "influx2":{
+       "addr": "http://192.168.100.245:8086",
+       "token": "test_token",
+       "org": "admin",
+       "measurement": "test",
+       "bucket": "bucketName",
+       "tagKey": "tagKey",
+       "tagValue": "tagValue",
+       "fields": "humidity,temperature,pressure"
+      }
+    }
+  ]
+}
+```
+### ####/tmp/influxPlugin.txt
+```json
+{
+  "file":"http://localhost:8080/influx2.zip"
+}
+```
+### plugins/go.mod
+```
+module plugins
+
+go 1.18
+
+require (
+        github.com/lf-edge/ekuiper v0.0.0-20220727015637-7d6f5c447110
+        github.com/influxdata/influxdb-client-go/v2 v2.10.0
+        github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839 // indirect
+)
+
+replace github.com/lf-edge/ekuiper => /root/goProject/ekuiper
+
+```
diff --git a/etc/functions/geohash.json b/etc/functions/geohash.json
deleted file mode 100644
index 06153e07..00000000
--- a/etc/functions/geohash.json
+++ /dev/null
@@ -1,92 +0,0 @@
-{
-	"about": {
-		"trial": false,
-		"author": {
-			"name": "EMQ",
-			"email": "contact@emqx.io",
-			"company": "EMQ Technologies Co., Ltd",
-			"website": "https://www.emqx.io"
-		},
-		"helpUrl": {
-			"en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/sqls/custom_functions.md",
-			"zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/sqls/custom_functions.md"
-		},
-		"description": {
-			"en_US": "",
-			"zh_CN": ""
-		}
-	},
-	"libs": ["github.com/mmcloughlin/geohash@master"],
-  "name":"geohash",
-	"functions": [{
-		"name": "geohashEncode",
-		"example": "geohashEncode(la,lo )",
-		"hint": {
-			"en_US": "Encode latitude and longitude as characters",
-			"zh_CN": "将经纬度编码为字符"
-		}
-	}, {
-		"name": "geohashEncodeInt",
-		"example": "geohashEncodeInt(la,lo )",
-		"hint": {
-			"en_US": "Encode latitude and longitude as numbers",
-			"zh_CN": "将经纬度编码为数字"
-		}
-	}, {
-		"name": "geohashDecode",
-		"example": "geohashDecode(hash )",
-		"hint": {
-			"en_US": "Decode characters into latitude and longitude",
-			"zh_CN": "将字符解码为经纬度"
-		}
-	}, {
-		"name": "geohashDecodeInt",
-		"example": "geohashDecodeInt(hash)",
-		"hint": {
-			"en_US": "Decode numbers into latitude and longitude",
-			"zh_CN": "将数字解码为经纬度"
-		}
-	}, {
-		"name": "geohashBoundingBox",
-		"example": "geohashBoundingBox(hash )",
-		"hint": {
-			"en_US": "Area for calculating character codes",
-			"zh_CN": "计算字符编码的区域"
-		}
-	}, {
-		"name": "geohashBoundingBoxInt",
-		"example": "geohashBoundingBoxInt(hash)",
-		"hint": {
-			"en_US": "Calculate the area of digital coding",
-			"zh_CN": "计算数字编码的区域"
-		}
-	}, {
-		"name": "geohashNeighbor",
-		"example": "geohashNeighbor(hash,direction )",
-		"hint": {
-			"en_US": "Calculate the neighbor of the corresponding direction of the character encoding",
-			"zh_CN": "计算字符编码对应方向的邻居"
-		}
-	}, {
-		"name": "geohashNeighborInt",
-		"example": "geohashNeighborInt(hash,direction )",
-		"hint": {
-			"en_US": "Calculate the neighbors in the corresponding direction of the digital code",
-			"zh_CN": "计算数字编码对应方向的邻居"
-		}
-	}, {
-		"name": "geohashNeighbors",
-		"example": "geohashNeighbors(hash)",
-		"hint": {
-			"en_US": "Calculate all neighbors of character encoding",
-			"zh_CN": "计算字符编码的所有邻居"
-		}
-	}, {
-		"name": "geohashNeighborsInt",
-		"example": "geohashNeighborsInt(hash)",
-		"hint": {
-			"en_US": "Calculate all neighbors of digital encoding",
-			"zh_CN": "计算数字编码的所有邻居"
-		}
-	}]
-}
diff --git a/etc/functions/image.json b/etc/functions/image.json
deleted file mode 100644
index 5e57d5d3..00000000
--- a/etc/functions/image.json
+++ /dev/null
@@ -1,36 +0,0 @@
-{
-	"about": {
-		"trial": false,
-		"author": {
-			"name": "EMQ",
-			"email": "contact@emqx.io",
-			"company": "EMQ Technologies Co., Ltd",
-			"website": "https://www.emqx.io"
-		},
-		"helpUrl": {
-			"en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/sqls/custom_functions.md",
-			"zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/sqls/custom_functions.md"
-		},
-		"description": {
-			"en_US": "",
-			"zh_CN": ""
-		}
-	},
-	"libs": ["github.com/nfnt/resize@master"],
-	"name": "image",
-	"functions": [{
-		"name": "resize",
-		"example": "resize(image,width, height)",
-		"hint": {
-			"en_US": "Creates a scaled image with new dimensions (width, height) .If either width or height is set to 0, it will be set to an aspect ratio preserving value.",
-			"zh_CN": "创建具有新尺寸（宽度，高度）的缩放图像。如果width或height设置为0，则将其设置为长宽比保留值。"
-		}
-	}, {
-		"name": "thumbnail",
-		"example": "thumbnail(image,maxWidth, maxHeight)",
-		"hint": {
-			"en_US": "Downscales an image preserving its aspect ratio to the maximum dimensions (maxWidth, maxHeight).",
-			"zh_CN": "将保留宽高比的图像缩小到最大尺寸（maxWidth，maxHeight）。"
-		}
-	}]
-}
diff --git a/etc/sinks/edgex.json b/etc/sinks/edgex.json
index 2fb51f8a..8c88fac7 100644
--- a/etc/sinks/edgex.json
+++ b/etc/sinks/edgex.json
@@ -416,5 +416,13 @@
         }
       ]
     }
-  ]
+  ],
+  "node": {
+    "category": "Sink",
+    "icon": "iconPath",
+    "label": {
+      "en": "EDGEX output",
+      "zh": "EDGEX 输出"
+    }
+  }
 }
diff --git a/etc/sinks/influx2.json b/etc/sinks/influx2.json
new file mode 100644
index 00000000..7bf3bb51
--- /dev/null
+++ b/etc/sinks/influx2.json
@@ -0,0 +1,135 @@
+{
+  "about": {
+    "trial": true,
+    "author": {
+      "name": "elpsyr",
+      "email": "hellccqcq@gmail.com",
+      "company": "",
+      "website": "https://github.com/elpsyr"
+    },
+    "helpUrl": {
+      "en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/rules/sinks/plugin/influx2.md",
+      "zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/rules/sinks/plugin/influx2.md"
+    },
+    "description": {
+      "en_US": "This a sink plugin for InfluxDB2, it can be used for saving the analysis data into InfluxDB V2.X .",
+      "zh_CN": "本插件为 InfluxDB2 的持久化插件，可以用于将分析数据存入 InfluxDB V2.X 中"
+    }
+  },
+  "libs": [
+    "github.com/influxdata/influxdb-client-go/v2"
+  ],
+  "properties": [{
+    "name": "addr",
+    "default": "http://192.168.100.245:8086",
+    "optional": false,
+    "control": "text",
+    "type": "string",
+    "hint": {
+      "en_US": "The addr of the InfluxDB",
+      "zh_CN": "InfluxDB的地址"
+    },
+    "label": {
+      "en_US": "Addr",
+      "zh_CN": "地址"
+    }
+  }, {
+    "name": "measurement",
+    "default": "test",
+    "optional": true,
+    "control": "text",
+    "type": "string",
+    "hint": {
+      "en_US": "The measurement of the InfluxDB",
+      "zh_CN": "InfluxDB的测量"
+    },
+    "label": {
+      "en_US": "Measurement",
+      "zh_CN": "测量"
+    }
+  }, {
+    "name": "token",
+    "default": "",
+    "optional": false,
+    "control": "text",
+    "type": "string",
+    "hint": {
+      "en_US": "The InfluxDB api token",
+      "zh_CN": "InfluxDB 验证 api token"
+    },
+    "label": {
+      "en_US": "API Token",
+      "zh_CN": "API Token"
+    }
+  }, {
+    "name": "org",
+    "default": "",
+    "optional": false,
+    "control": "text",
+    "type": "string",
+    "hint": {
+      "en_US": "The InfluxDB org",
+      "zh_CN": "InfluxDB org"
+    },
+    "label": {
+      "en_US": "org",
+      "zh_CN": "组织"
+    }
+  }, {
+    "name": "bucket",
+    "default": "",
+    "optional": false,
+    "control": "text",
+    "type": "string",
+    "hint": {
+      "en_US": "The InfluxDB bucket",
+      "zh_CN": "InfluxDB bucket"
+    },
+    "label": {
+      "en_US": "bucket",
+      "zh_CN": "桶"
+    }
+  }, {
+    "name": "tagKey",
+    "default": "tagKey",
+    "optional": true,
+    "control": "text",
+    "type": "string",
+    "hint": {
+      "en_US": "The tag key of the InfluxDB",
+      "zh_CN": "InfluxDB 的标签键"
+    },
+    "label": {
+      "en_US": "Tag",
+      "zh_CN": "标签键"
+    }
+  }, {
+    "name": "tagValue",
+    "default": "tagValue",
+    "optional": true,
+    "control": "text",
+    "type": "string",
+    "hint": {
+      "en_US": "The tag value of the InfluxDB",
+      "zh_CN": "InfluxDB 的标签值"
+    },
+    "label": {
+      "en_US": "Tag value",
+      "zh_CN": "标签值"
+    }
+  }, {
+    "name": "fields",
+    "default": "humidity,temperature,pressure",
+    "optional": true,
+    "control": "text",
+    "type": "string",
+    "hint": {
+      "en_US": "The column of the InfluxDB",
+      "zh_CN": "InfluxDB 的列名"
+    },
+    "label": {
+      "en_US": "Column",
+      "zh_CN": "列名"
+    }
+  }]
+}
diff --git a/etc/sinks/log.json b/etc/sinks/log.json
index 5fdd87bf..9a2eba17 100644
--- a/etc/sinks/log.json
+++ b/etc/sinks/log.json
@@ -16,6 +16,13 @@
       "zh_CN": "该操作用于将输出消息打印到日志文件中，默认情况下，该日志文件位于 $ekuiper_install/log/stream.log 中。"
     }
   },
-  "properties": [
-  ]
+  "properties": [],
+  "node": {
+    "category": "Sink",
+    "icon": "iconPath",
+    "label": {
+      "en": "Log output",
+      "zh": "Log 输出"
+    }
+  }
 }
\ No newline at end of file
diff --git a/etc/sinks/memory.json b/etc/sinks/memory.json
index 8b4efdfa..d7aa6b60 100644
--- a/etc/sinks/memory.json
+++ b/etc/sinks/memory.json
@@ -32,5 +32,13 @@
         "zh_CN": "主题"
       }
     }
-  ]
+  ],
+  "node": {
+    "category": "Sink",
+    "icon": "iconPath",
+    "label": {
+      "en": "Memory output",
+      "zh": "内存输出"
+    }
+  }
 }
\ No newline at end of file
diff --git a/etc/sinks/mqtt.json b/etc/sinks/mqtt.json
index 818eec62..c875234a 100644
--- a/etc/sinks/mqtt.json
+++ b/etc/sinks/mqtt.json
@@ -216,5 +216,13 @@
         "zh_CN": "跳过证书验证"
       }
     }
-  ]
+  ],
+  "node": {
+    "category": "Sink",
+    "icon": "iconPath",
+    "label": {
+      "en": "MQTT output",
+      "zh": "MQTT 输出"
+    }
+  }
 }
diff --git a/etc/sinks/neuron.json b/etc/sinks/neuron.json
index e8e05d03..9dc8ef1c 100644
--- a/etc/sinks/neuron.json
+++ b/etc/sinks/neuron.json
@@ -77,5 +77,13 @@
         "zh_CN": "原始字符串"
       }
     }
-  ]
+  ],
+  "node": {
+    "category": "Sink",
+    "icon": "iconPath",
+    "label": {
+      "en": "Neuron output",
+      "zh": "Neuron 输出"
+    }
+  }
 }
\ No newline at end of file
diff --git a/etc/sinks/nop.json b/etc/sinks/nop.json
index 5994d1a9..f0c9d4f2 100644
--- a/etc/sinks/nop.json
+++ b/etc/sinks/nop.json
@@ -32,5 +32,13 @@
         "zh_CN": "打印日志"
       }
     }
-  ]
+  ],
+  "node": {
+    "category": "Sink",
+    "icon": "iconPath",
+    "label": {
+      "en": "Nop output",
+      "zh": "Nop 输出"
+    }
+  }
 }
\ No newline at end of file
diff --git a/etc/sinks/rest.json b/etc/sinks/rest.json
index e98e2454..04b860bd 100644
--- a/etc/sinks/rest.json
+++ b/etc/sinks/rest.json
@@ -186,5 +186,13 @@
         "zh_CN": "打印 HTTP 响应"
       }
     }
-  ]
+  ],
+  "node": {
+    "category": "Sink",
+    "icon": "iconPath",
+    "label": {
+      "en": "Rest output",
+      "zh": "rest 输出"
+    }
+  }
 }
diff --git a/etc/sources/edgex.json b/etc/sources/edgex.json
index 43c46908..6633ee29 100644
--- a/etc/sources/edgex.json
+++ b/etc/sources/edgex.json
@@ -339,5 +339,22 @@
 				}
 			}
 		]
+	},
+	"outputs": [
+		{
+			"label": {
+				"en_US": "Output",
+				"zh_CN": "输出"
+			},
+			"value": "signal"
+		}
+	],
+	"node": {
+		"category": "Protocol/Protocol1",
+		"icon": "iconPath",
+		"label": {
+			"en_US": "Edgex",
+			"zh_CN": "Edgex"
+		}
 	}
 }
diff --git a/etc/sources/file.json b/etc/sources/file.json
index d2ef9650..d9589e7e 100644
--- a/etc/sources/file.json
+++ b/etc/sources/file.json
@@ -62,5 +62,22 @@
           "zh_CN": "间隔时间"
         }
       }]
+  },
+  "outputs": [
+    {
+      "label": {
+        "en_US": "Output",
+        "zh_CN": "输出"
+      },
+      "value": "signal"
+    }
+  ],
+  "node": {
+    "category": "Protocol/Protocol1",
+    "icon": "iconPath",
+    "label": {
+      "en_US": "File",
+      "zh_CN": "File"
+    }
   }
 }
diff --git a/etc/sources/httppull.json b/etc/sources/httppull.json
index 0cde0e37..449de1ed 100644
--- a/etc/sources/httppull.json
+++ b/etc/sources/httppull.json
@@ -205,5 +205,22 @@
 				"zh_CN": "HTTP标头"
 			}
 		}]
+	},
+	"outputs": [
+		{
+			"label": {
+				"en_US": "Output",
+				"zh_CN": "输出"
+			},
+			"value": "signal"
+		}
+	],
+	"node": {
+		"category": "Protocol/Protocol1",
+		"icon": "iconPath",
+		"label": {
+			"en_US": "HTTP PULL",
+			"zh_CN": "HTTP PULL"
+		}
 	}
 }
diff --git a/etc/sources/httppush.json b/etc/sources/httppush.json
index 92ea4a49..522b4671 100644
--- a/etc/sources/httppush.json
+++ b/etc/sources/httppush.json
@@ -34,5 +34,20 @@
 				"zh_CN": "请求方法"
 			}
 		}]
+	},
+	"outputs": [{
+		"label": {
+			"en_US": "Output",
+			"zh_CN": "输出"
+		},
+		"value": "signal"
+	}],
+	"node": {
+		"category": "Protocol/Protocol1",
+		"icon": "iconPath",
+		"label": {
+			"en_US": "HTTP PUSH",
+			"zh_CN": "HTTP PUSH"
+		}
 	}
 }
diff --git a/etc/sources/memory.json b/etc/sources/memory.json
index 241d35dd..ae55c719 100644
--- a/etc/sources/memory.json
+++ b/etc/sources/memory.json
@@ -18,5 +18,22 @@
   },
   "libs": [],
   "properties": {
+  },
+  "outputs": [
+    {
+      "label": {
+        "en_US": "Output",
+        "zh_CN": "输出"
+      },
+      "value": "signal"
+    }
+  ],
+  "node": {
+    "category": "Protocol/Protocol1",
+    "icon": "iconPath",
+    "label": {
+      "en_US": "Memory",
+      "zh_CN": "Memory"
+    }
   }
 }
diff --git a/etc/sources/neuron.json b/etc/sources/neuron.json
index 47eaaff2..b0308146 100644
--- a/etc/sources/neuron.json
+++ b/etc/sources/neuron.json
@@ -17,6 +17,22 @@
     }
   },
   "libs": [],
-  "properties": {
+  "properties": {},
+  "outputs": [
+    {
+      "label": {
+        "en_US": "Output",
+        "zh_CN": "输出"
+      },
+      "value": "signal"
+    }
+  ],
+  "node": {
+    "category": "Protocol/Protocol1",
+    "icon": "iconPath",
+    "label": {
+      "en_US": "Neuron",
+      "zh_CN": "Neuron"
+    }
   }
 }
diff --git a/etc/functions/accumulateWordCount.json b/extensions/functions/accumulateWordCount/accumulateWordCount.json
similarity index 52%
rename from etc/functions/accumulateWordCount.json
rename to extensions/functions/accumulateWordCount/accumulateWordCount.json
index 8478be11..ba2afb69 100644
--- a/etc/functions/accumulateWordCount.json
+++ b/extensions/functions/accumulateWordCount/accumulateWordCount.json
@@ -19,9 +19,45 @@
 	"functions": [{
 		"name": "accumulateWordCount",
 		"example": "accumulateWordCount(col1)",
+		"aggregate": false,
 		"hint": {
 			"en_US": "This function counts the total number of words",
 			"zh_CN": "该函数统计一共有多少个单词"
+		},
+		"args": [
+			{
+				"name": "inputData",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				},
+				"label": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				},
+				"values": []
+			}
+		],
+		"outputs": [
+			{
+				"label": {
+					"en_US": "Output",
+					"zh_CN": "输出"
+				},
+				"value": "accumulateWordCount"
+			}
+		],
+		"node": {
+			"category": "abc",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "AccumulateWordCount",
+				"zh_CN": "AccumulateWordCount"
+			}
 		}
 	}]
 }
diff --git a/etc/functions/countPlusOne.json b/extensions/functions/countPlusOne/countPlusOne.json
similarity index 54%
rename from etc/functions/countPlusOne.json
rename to extensions/functions/countPlusOne/countPlusOne.json
index 3a9ec815..d08644c3 100644
--- a/etc/functions/countPlusOne.json
+++ b/extensions/functions/countPlusOne/countPlusOne.json
@@ -22,6 +22,40 @@
 		"hint": {
 			"en_US": "The value of the output parameter length plus one",
 			"zh_CN": "输出参数长度加一的值"
+		},
+		"args": [
+			{
+				"name": "inputData",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				},
+				"label": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				}
+			}
+		],
+		"outputs": [
+			{
+				"label": {
+					"en_US": "Output",
+					"zh_CN": "输出"
+				},
+				"value": "countPlusOne"
+			}
+		],
+		"node": {
+			"category": "abc",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "countPlusOne",
+				"zh_CN": "countPlusOne"
+			}
 		}
 	}]
 }
diff --git a/etc/functions/echo.json b/extensions/functions/echo/echo.json
similarity index 53%
rename from etc/functions/echo.json
rename to extensions/functions/echo/echo.json
index 68f6ab14..7ad15697 100644
--- a/etc/functions/echo.json
+++ b/extensions/functions/echo/echo.json
@@ -22,6 +22,40 @@
 		"hint": {
 			"en_US": "The parameter value is output as it is.",
 			"zh_CN": "原样输出参数值。"
+		},
+		"args": [
+			{
+				"name": "inputData",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				},
+				"label": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				}
+			}
+		],
+		"outputs": [
+			{
+				"label": {
+					"en_US": "Output",
+					"zh_CN": "输出"
+				},
+				"value": "echo"
+			}
+		],
+		"node": {
+			"category": "abc",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "countPlusOne",
+				"zh_CN": "countPlusOne"
+			}
 		}
 	}]
 }
diff --git a/extensions/functions/geohash/geohash.json b/extensions/functions/geohash/geohash.json
new file mode 100644
index 00000000..af2462ae
--- /dev/null
+++ b/extensions/functions/geohash/geohash.json
@@ -0,0 +1,488 @@
+{
+	"about": {
+		"trial": false,
+		"author": {
+			"name": "EMQ",
+			"email": "contact@emqx.io",
+			"company": "EMQ Technologies Co., Ltd",
+			"website": "https://www.emqx.io"
+		},
+		"helpUrl": {
+			"en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/sqls/custom_functions.md",
+			"zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/sqls/custom_functions.md"
+		},
+		"description": {
+			"en_US": "",
+			"zh_CN": ""
+		}
+	},
+	"libs": ["github.com/mmcloughlin/geohash@master"],
+  "name":"geohash",
+	"functions": [{
+		"name": "geohashEncode",
+		"example": "geohashEncode(la,lo)",
+		"hint": {
+			"en_US": "Encode latitude and longitude as characters",
+			"zh_CN": "将经纬度编码为字符"
+		},
+		"args": [
+			{
+				"name": "la",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input la",
+					"zh_CN": "输入经度"
+				},
+				"label": {
+					"en_US": "Input la",
+					"zh_CN": "输入经度"
+				}
+			},
+			{
+				"name": "lo",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input lo",
+					"zh_CN": "输入纬度"
+				},
+				"label": {
+					"en_US": "Input lo",
+					"zh_CN": "输入纬度"
+				}
+			}
+		],
+		"outputs": [
+			{
+				"label": {
+					"en_US": "Output",
+					"zh_CN": "输出"
+				},
+				"value": "geohashEncode"
+			}
+		],
+		"node": {
+			"category": "abc",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "geohashEncode",
+				"zh_CN": "geohashEncode"
+			}
+		}
+	}, {
+		"name": "geohashEncodeInt",
+		"example": "geohashEncodeInt(la,lo )",
+		"hint": {
+			"en_US": "Encode latitude and longitude as numbers",
+			"zh_CN": "将经纬度编码为数字"
+		},
+		"args": [
+			{
+				"name": "la",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input la",
+					"zh_CN": "输入经度"
+				},
+				"label": {
+					"en_US": "Input la",
+					"zh_CN": "输入经度"
+				}
+			},
+			{
+				"name": "lo",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input lo",
+					"zh_CN": "输入纬度"
+				},
+				"label": {
+					"en_US": "Input lo",
+					"zh_CN": "输入纬度"
+				}
+			}
+		],
+		"outputs": [
+			{
+				"label": {
+					"en_US": "Output",
+					"zh_CN": "输出"
+				},
+				"value": "geohashEncodeInt"
+			}
+		],
+		"node": {
+			"category": "abc",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "geohashEncodeInt",
+				"zh_CN": "geohashEncodeInt"
+			}
+		}
+	}, {
+		"name": "geohashDecode",
+		"example": "geohashDecode(hash )",
+		"hint": {
+			"en_US": "Decode characters into latitude and longitude",
+			"zh_CN": "将字符解码为经纬度"
+		},
+		"args": [
+			{
+				"name": "data",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				},
+				"label": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				}
+			}
+		],
+		"outputs": [
+			{
+				"label": {
+					"en_US": "Output",
+					"zh_CN": "输出"
+				},
+				"value": "geohashDecode"
+			}
+		],
+		"node": {
+			"category": "abc",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "geohashDecode",
+				"zh_CN": "geohashDecode"
+			}
+		}
+	}, {
+		"name": "geohashDecodeInt",
+		"example": "geohashDecodeInt(hash)",
+		"hint": {
+			"en_US": "Decode numbers into latitude and longitude",
+			"zh_CN": "将数字解码为经纬度"
+		},
+		"args": [
+			{
+				"name": "data",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				},
+				"label": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				}
+			}
+		],
+		"outputs": [
+			{
+				"label": {
+					"en_US": "Output",
+					"zh_CN": "输出"
+				},
+				"value": "geohashDecodeInt"
+			}
+		],
+		"node": {
+			"category": "abc",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "geohashDecodeInt",
+				"zh_CN": "geohashDecodeInt"
+			}
+		}
+	}, {
+		"name": "geohashBoundingBox",
+		"example": "geohashBoundingBox(hash )",
+		"hint": {
+			"en_US": "Area for calculating character codes",
+			"zh_CN": "计算字符编码的区域"
+		},
+		"args": [
+			{
+				"name": "data",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				},
+				"label": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				}
+			}
+		],
+		"outputs": [
+			{
+				"label": {
+					"en_US": "Output",
+					"zh_CN": "输出"
+				},
+				"value": "geohashBoundingBox"
+			}
+		],
+		"node": {
+			"category": "abc",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "geohashBoundingBox",
+				"zh_CN": "geohashBoundingBox"
+			}
+		}
+	}, {
+		"name": "geohashBoundingBoxInt",
+		"example": "geohashBoundingBoxInt(hash)",
+		"hint": {
+			"en_US": "Calculate the area of digital coding",
+			"zh_CN": "计算数字编码的区域"
+		},
+		"args": [
+			{
+				"name": "data",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				},
+				"label": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				}
+			}
+		],
+		"outputs": [
+			{
+				"label": {
+					"en_US": "Output",
+					"zh_CN": "输出"
+				},
+				"value": "geohashBoundingBoxInt"
+			}
+		],
+		"node": {
+			"category": "abc",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "geohashBoundingBoxInt",
+				"zh_CN": "geohashBoundingBoxInt"
+			}
+		}
+	}, {
+		"name": "geohashNeighbor",
+		"example": "geohashNeighbor(hash,direction )",
+		"hint": {
+			"en_US": "Calculate the neighbor of the corresponding direction of the character encoding",
+			"zh_CN": "计算字符编码对应方向的邻居"
+		},
+		"args": [
+			{
+				"name": "hash",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				},
+				"label": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				}
+			},
+			{
+				"name": "direction",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input direction",
+					"zh_CN": "输入数据"
+				},
+				"label": {
+					"en_US": "Input direction",
+					"zh_CN": "输入数据"
+				}
+			}
+		],
+		"outputs": [
+			{
+				"label": {
+					"en_US": "Output",
+					"zh_CN": "输出"
+				},
+				"value": "geohashNeighbor"
+			}
+		],
+		"node": {
+			"category": "abc",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "geohashNeighbor",
+				"zh_CN": "geohashNeighbor"
+			}
+		}
+	}, {
+		"name": "geohashNeighborInt",
+		"example": "geohashNeighborInt(hash,direction )",
+		"hint": {
+			"en_US": "Calculate the neighbors in the corresponding direction of the digital code",
+			"zh_CN": "计算数字编码对应方向的邻居"
+		},
+		"args": [
+			{
+				"name": "hash",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				},
+				"label": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				}
+			},
+			{
+				"name": "direction",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input direction",
+					"zh_CN": "输入数据"
+				},
+				"label": {
+					"en_US": "Input direction",
+					"zh_CN": "输入数据"
+				}
+			}
+		],
+		"outputs": [
+			{
+				"label": {
+					"en_US": "Output",
+					"zh_CN": "输出"
+				},
+				"value": "geohashNeighborInt"
+			}
+		],
+		"node": {
+			"category": "abc",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "geohashNeighborInt",
+				"zh_CN": "geohashNeighborInt"
+			}
+		}
+	}, {
+		"name": "geohashNeighbors",
+		"example": "geohashNeighbors(hash)",
+		"hint": {
+			"en_US": "Calculate all neighbors of character encoding",
+			"zh_CN": "计算字符编码的所有邻居"
+		},
+		"args": [
+			{
+				"name": "hash",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				},
+				"label": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				}
+			}
+		],
+		"outputs": [
+			{
+				"label": {
+					"en_US": "Output",
+					"zh_CN": "输出"
+				},
+				"value": "geohashNeighbors"
+			}
+		],
+		"node": {
+			"category": "abc",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "geohashNeighbors",
+				"zh_CN": "geohashNeighbors"
+			}
+		}
+	}, {
+		"name": "geohashNeighborsInt",
+		"example": "geohashNeighborsInt(hash)",
+		"hint": {
+			"en_US": "Calculate all neighbors of digital encoding",
+			"zh_CN": "计算数字编码的所有邻居"
+		},
+		"args": [{
+				"name": "hash",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				},
+				"label": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				}
+			}],
+		"outputs": [{
+			"label": {
+				"en_US": "Output",
+				"zh_CN": "输出"
+			},
+			"value": "geohashNeighbors"
+		}],
+		"node": {
+				"category": "abc",
+				"icon": "iconPath",
+				"label": {
+					"en_US": "geohashNeighbors",
+					"zh_CN": "geohashNeighbors"
+				}
+			}
+	}]
+}
diff --git a/extensions/functions/image/image.json b/extensions/functions/image/image.json
new file mode 100644
index 00000000..a55d71d0
--- /dev/null
+++ b/extensions/functions/image/image.json
@@ -0,0 +1,160 @@
+{
+	"about": {
+		"trial": false,
+		"author": {
+			"name": "EMQ",
+			"email": "contact@emqx.io",
+			"company": "EMQ Technologies Co., Ltd",
+			"website": "https://www.emqx.io"
+		},
+		"helpUrl": {
+			"en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/sqls/custom_functions.md",
+			"zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/sqls/custom_functions.md"
+		},
+		"description": {
+			"en_US": "",
+			"zh_CN": ""
+		}
+	},
+	"libs": ["github.com/nfnt/resize@master"],
+	"name": "image",
+	"functions": [{
+		"name": "resize",
+		"example": "resize(image,width, height)",
+		"hint": {
+			"en_US": "Creates a scaled image with new dimensions (width, height) .If either width or height is set to 0, it will be set to an aspect ratio preserving value.",
+			"zh_CN": "创建具有新尺寸（宽度，高度）的缩放图像。如果width或height设置为0，则将其设置为长宽比保留值。"
+		},
+		"args": [
+			{
+				"name": "image",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				},
+				"label": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				}
+			},{
+				"name": "width",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "Input width",
+					"zh_CN": "输入宽度"
+				},
+				"label": {
+					"en_US": "Input width",
+					"zh_CN": "输入宽度"
+				}
+			},{
+				"name": "height",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "Input height",
+					"zh_CN": "输入高度"
+				},
+				"label": {
+					"en_US": "Input height",
+					"zh_CN": "输入高度"
+				}
+			}
+		],
+		"outputs": [
+			{
+				"label": {
+					"en_US": "Output",
+					"zh_CN": "输出"
+				},
+				"value": "resize"
+			}
+		],
+		"node": {
+			"category": "abc",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "resize",
+				"zh_CN": "resize"
+			}
+		}
+	}, {
+		"name": "thumbnail",
+		"example": "thumbnail(image,maxWidth, maxHeight)",
+		"hint": {
+			"en_US": "Downscales an image preserving its aspect ratio to the maximum dimensions (maxWidth, maxHeight).",
+			"zh_CN": "将保留宽高比的图像缩小到最大尺寸（maxWidth，maxHeight）。"
+		},
+		"args": [
+			{
+				"name": "image",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				},
+				"label": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				}
+			},{
+				"name": "maxWidth",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "Input maxWidth",
+					"zh_CN": "输入最大宽度"
+				},
+				"label": {
+					"en_US": "Input maxWidth",
+					"zh_CN": "输入最大宽度"
+				}
+			},{
+				"name": "maxHeight",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "Input maxHeight",
+					"zh_CN": "输入最大高度"
+				},
+				"label": {
+					"en_US": "Input maxHeight",
+					"zh_CN": "输入最大高度"
+				}
+			}
+		],
+		"outputs": [
+			{
+				"label": {
+					"en_US": "Output",
+					"zh_CN": "输出"
+				},
+				"value": "thumbnail"
+			}
+		],
+		"node": {
+			"category": "abc",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "thumbnail",
+				"zh_CN": "thumbnail"
+			}
+		}
+	}]
+}
diff --git a/etc/functions/labelImage.json b/extensions/functions/labelImage/labelImage.json
similarity index 54%
rename from etc/functions/labelImage.json
rename to extensions/functions/labelImage/labelImage.json
index 71f421d7..b2bab852 100644
--- a/etc/functions/labelImage.json
+++ b/extensions/functions/labelImage/labelImage.json
@@ -22,6 +22,40 @@
 		"hint": {
 			"en_US": "Label an image by tensorflow lite model.",
 			"zh_CN": "采用 tensorflow lite 模型标记图片。"
+		},
+		"args": [
+			{
+				"name": "inputData",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				},
+				"label": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据"
+				}
+			}
+		],
+		"outputs": [
+			{
+				"label": {
+					"en_US": "Output",
+					"zh_CN": "输出"
+				},
+				"value": "labelImage"
+			}
+		],
+		"node": {
+			"category": "abc",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "labelImage",
+				"zh_CN": "labelImage"
+			}
 		}
 	}]
 }
diff --git a/etc/sinks/file.json b/extensions/sinks/file/file.json
similarity index 91%
rename from etc/sinks/file.json
rename to extensions/sinks/file/file.json
index cb7dcbf6..8ea483fd 100644
--- a/etc/sinks/file.json
+++ b/extensions/sinks/file/file.json
@@ -46,5 +46,13 @@
 			"en_US": "Intervals",
 			"zh_CN": "间隔时间"
 		}
-	}]
+	}],
+	"node": {
+		"category": "Sink",
+		"icon": "iconPath",
+		"label": {
+			"en": "File",
+			"zh": "文件"
+		}
+	}
 }
diff --git a/extensions/sinks/image/image.go b/extensions/sinks/image/image.go
index 58adb617..c4dc0439 100644
--- a/extensions/sinks/image/image.go
+++ b/extensions/sinks/image/image.go
@@ -37,7 +37,7 @@ type imageSink struct {
 }
 
 func (m *imageSink) Configure(props map[string]interface{}) error {
-	if i, ok := props["format"]; ok {
+	if i, ok := props["imageFormat"]; ok {
 		if i, ok := i.(string); ok {
 			if "png" != i && "jpeg" != i {
 				return fmt.Errorf("%s image type is not currently supported", i)
diff --git a/extensions/sinks/image/image.json b/extensions/sinks/image/image.json
new file mode 100644
index 00000000..22b50392
--- /dev/null
+++ b/extensions/sinks/image/image.json
@@ -0,0 +1,82 @@
+{
+	"about": {
+		"trial": true,
+		"author": {
+			"name": "EMQ",
+			"email": "contact@emqx.io",
+			"company": "EMQ Technologies Co., Ltd",
+			"website": "https://www.emqx.io"
+		},
+		"helpUrl": {
+			"en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/rules/sinks/plugin/file.md",
+			"zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/rules/sinks/plugin/file.md"
+		},
+		"description": {
+			"en_US": "This a sink plugin for file, it can be used for saving the analysis data into file system.",
+			"zh_CN": "本插件为文件持久化插件，可以用于将分析数据存入指定的文件中。"
+		}
+	},
+	"libs": [
+	],
+	"properties": [{
+		"name": "path",
+		"default": "",
+		"optional": false,
+		"control": "text",
+		"type": "string",
+		"hint": {
+			"en_US": "The name of the folder where the pictures are saved, such as ./tmp. Note: For multiple rules, their paths cannot be repeated, otherwise they will be deleted from each other.",
+			"zh_CN": "保存图片的文件夹名，例如 ./tmp。注意：多条 rule 路径不能重复，否则会出现彼此删除的现象。"
+		},
+		"label": {
+			"en_US": "Path of file",
+			"zh_CN": "文件路径"
+		}
+	}, {
+		"name": "imageFormat",
+		"default": "jpeg",
+		"optional": true,
+		"control": "select",
+		"values": [
+			"jpeg",
+			"png"
+		],
+		"type": "list_string",
+		"hint": {
+			"en_US": "File format, support jpeg and png.",
+			"zh_CN": "文件格式，支持 jpeg 和 png。"
+		},
+		"label": {
+			"en_US": "The format of image",
+			"zh_CN": "图片格式"
+		}
+	},{
+		"name": "maxAge",
+		"default": 1000,
+		"optional": true,
+		"control": "text",
+		"type": "int",
+		"hint": {
+			"en_US": "Maximum file storage time (hours). The default value is 72, which means that the picture can be stored for up to 3 days.",
+			"zh_CN": "最长文件存储时间(小时)。默认值为72，这表示图片最多保存3天。"
+		},
+		"label": {
+			"en_US": "maxAge",
+			"zh_CN": "最长保留时间"
+		}
+	},{
+		"name": "maxCount",
+		"default": 1000,
+		"optional": true,
+		"control": "text",
+		"type": "int",
+		"hint": {
+			"en_US": "The maximum number of stored pictures. The default value is 1000. The earlier pictures will be deleted. The relationship with maxAge is OR.",
+			"zh_CN": "存储图片的最大数量，默认值是1000，删除时间较早的图片，与maxAge是或的关系。"
+		},
+		"label": {
+			"en_US": "maxCount",
+			"zh_CN": "最大写入数量"
+		}
+	}]
+}
diff --git a/etc/sinks/influx.json b/extensions/sinks/influx/influx.json
similarity index 96%
rename from etc/sinks/influx.json
rename to extensions/sinks/influx/influx.json
index 5ea77c7c..eee48677 100644
--- a/etc/sinks/influx.json
+++ b/extensions/sinks/influx/influx.json
@@ -131,5 +131,13 @@
 			"en_US": "Column",
 			"zh_CN": "列名"
 		}
-	}]
+	}],
+	"node": {
+		"category": "Sink",
+		"icon": "iconPath",
+		"label": {
+			"en": "Influx",
+			"zh": "Influx"
+		}
+	}
 }
diff --git a/extensions/sinks/influx2/influx2.go b/extensions/sinks/influx2/influx2.go
new file mode 100644
index 00000000..86540420
--- /dev/null
+++ b/extensions/sinks/influx2/influx2.go
@@ -0,0 +1,169 @@
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+//
+// 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.
+
+// 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.
+
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	_ "github.com/influxdata/influxdb-client-go/v2"
+	client "github.com/influxdata/influxdb-client-go/v2"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"strings"
+	"time"
+)
+
+type influxSink2 struct {
+	addr         string
+	token        string
+	measurement  string
+	org          string
+	bucket       string
+	tagKey       string
+	tagValue     string
+	fields       string
+	cli          client.Client
+	fieldMap     map[string]interface{}
+	hasTransform bool
+}
+
+func (m *influxSink2) Configure(props map[string]interface{}) error {
+	if i, ok := props["addr"]; ok {
+		if i, ok := i.(string); ok {
+			m.addr = i
+		}
+	}
+	if i, ok := props["measurement"]; ok {
+		if i, ok := i.(string); ok {
+			m.measurement = i
+		}
+	}
+	if i, ok := props["tagKey"]; ok {
+		if i, ok := i.(string); ok {
+			m.tagKey = i
+		}
+	}
+	if i, ok := props["tagValue"]; ok {
+		if i, ok := i.(string); ok {
+			m.tagValue = i
+		}
+	}
+	if i, ok := props["fields"]; ok {
+		if i, ok := i.(string); ok {
+			m.fields = i
+		}
+	}
+	if i, ok := props["dataTemplate"]; ok {
+		if i, ok := i.(string); ok && i != "" {
+			m.hasTransform = true
+		}
+	}
+
+	if i, ok := props["token"]; ok {
+		if i, ok := i.(string); ok {
+			m.token = i
+		}
+	}
+	if i, ok := props["org"]; ok {
+		if i, ok := i.(string); ok {
+			m.org = i
+		}
+	}
+	if i, ok := props["bucket"]; ok {
+		if i, ok := i.(string); ok {
+			m.bucket = i
+		}
+	}
+
+	return nil
+}
+
+func (m *influxSink2) Open(ctx api.StreamContext) (err error) {
+	logger := ctx.GetLogger()
+	logger.Debug("Opening influx2 sink")
+	options := client.DefaultOptions().SetBatchSize(100)
+	m.cli = client.NewClientWithOptions(m.addr, m.token, options)
+	return nil
+}
+
+func (m *influxSink2) Collect(ctx api.StreamContext, data interface{}) error {
+	logger := ctx.GetLogger()
+	if m.hasTransform {
+		jsonBytes, _, err := ctx.TransformOutput(data)
+		if err != nil {
+			return err
+		}
+		m := make(map[string]interface{})
+		err = json.Unmarshal(jsonBytes, &m)
+		if err != nil {
+			return fmt.Errorf("fail to decode data %s after applying dataTemplate for error %v", string(jsonBytes), err)
+		}
+		data = m
+	}
+	var output map[string]interface{}
+	switch v := data.(type) {
+	case map[string]interface{}:
+		output = v
+	case []map[string]interface{}:
+		if len(v) > 0 {
+			output = v[0]
+		} else {
+			ctx.GetLogger().Warnf("Get empty data %v, just return", data)
+			return nil
+		}
+	}
+
+	writeAPI := m.cli.WriteAPIBlocking(m.org, m.bucket)
+
+	tags := map[string]string{m.tagKey: m.tagValue}
+	fields := strings.Split(m.fields, ",")
+	m.fieldMap = make(map[string]interface{}, 100)
+	for _, field := range fields {
+		if output[field] != nil {
+			m.fieldMap[field] = output[field]
+		}
+	}
+
+	pt := client.NewPoint(m.measurement, tags, m.fieldMap, time.Now())
+
+	err := writeAPI.WritePoint(ctx, pt)
+	if err != nil {
+		logger.Debug(err)
+		return err
+	}
+	logger.Debug("insert data into influxdb2 success")
+
+	return nil
+}
+
+func (m *influxSink2) Close(ctx api.StreamContext) error {
+	m.cli.Close()
+	return nil
+}
+
+func Influx2() api.Sink {
+	return &influxSink2{}
+}
diff --git a/extensions/sinks/influx2/influx2.json b/extensions/sinks/influx2/influx2.json
new file mode 100644
index 00000000..dfe0106d
--- /dev/null
+++ b/extensions/sinks/influx2/influx2.json
@@ -0,0 +1,135 @@
+{
+	"about": {
+		"trial": true,
+		"author": {
+			"name": "elpsyr",
+			"email": "hellccqcq@gmail.com",
+			"company": "",
+			"website": "https://github.com/elpsyr"
+		},
+		"helpUrl": {
+			"en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/rules/sinks/plugin/influx2.md",
+			"zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/rules/sinks/plugin/influx2.md"
+		},
+		"description": {
+			"en_US": "This a sink plugin for InfluxDB2, it can be used for saving the analysis data into InfluxDB V2.X .",
+			"zh_CN": "本插件为 InfluxDB2 的持久化插件，可以用于将分析数据存入 InfluxDB V2.X 中"
+		}
+	},
+	"libs": [
+	  "github.com/influxdata/influxdb-client-go/v2@master"
+	],
+	"properties": [{
+		"name": "addr",
+		"default": "http://192.168.100.245:8086",
+		"optional": false,
+		"control": "text",
+		"type": "string",
+		"hint": {
+			"en_US": "The addr of the InfluxDB",
+			"zh_CN": "InfluxDB的地址"
+		},
+		"label": {
+			"en_US": "Addr",
+			"zh_CN": "地址"
+		}
+	}, {
+		"name": "measurement",
+		"default": "test",
+		"optional": true,
+		"control": "text",
+		"type": "string",
+		"hint": {
+			"en_US": "The measurement of the InfluxDB",
+			"zh_CN": "InfluxDB的测量"
+		},
+		"label": {
+			"en_US": "Measurement",
+			"zh_CN": "测量"
+		}
+	}, {
+		"name": "token",
+		"default": "",
+		"optional": false,
+		"control": "text",
+		"type": "string",
+		"hint": {
+			"en_US": "The InfluxDB api token",
+			"zh_CN": "InfluxDB 验证 api token"
+		},
+		"label": {
+			"en_US": "API Token",
+			"zh_CN": "API Token"
+		}
+	}, {
+		"name": "org",
+		"default": "",
+		"optional": false,
+		"control": "text",
+		"type": "string",
+		"hint": {
+			"en_US": "The InfluxDB org",
+			"zh_CN": "InfluxDB org"
+		},
+		"label": {
+			"en_US": "org",
+			"zh_CN": "组织"
+		}
+	}, {
+		"name": "bucket",
+		"default": "",
+		"optional": false,
+		"control": "text",
+		"type": "string",
+		"hint": {
+			"en_US": "The InfluxDB bucket",
+			"zh_CN": "InfluxDB bucket"
+		},
+		"label": {
+			"en_US": "bucket",
+			"zh_CN": "桶"
+		}
+	}, {
+		"name": "tagKey",
+		"default": "tagKey",
+		"optional": true,
+		"control": "text",
+		"type": "string",
+		"hint": {
+			"en_US": "The tag key of the InfluxDB",
+			"zh_CN": "InfluxDB 的标签键"
+		},
+		"label": {
+			"en_US": "Tag",
+			"zh_CN": "标签键"
+		}
+	}, {
+		"name": "tagValue",
+		"default": "tagValue",
+		"optional": true,
+		"control": "text",
+		"type": "string",
+		"hint": {
+			"en_US": "The tag value of the InfluxDB",
+			"zh_CN": "InfluxDB 的标签值"
+		},
+		"label": {
+			"en_US": "Tag value",
+			"zh_CN": "标签值"
+		}
+	}, {
+		"name": "fields",
+		"default": "humidity,temperature,pressure",
+		"optional": true,
+		"control": "text",
+		"type": "string",
+		"hint": {
+			"en_US": "The column of the InfluxDB",
+			"zh_CN": "InfluxDB 的列名"
+		},
+		"label": {
+			"en_US": "Column",
+			"zh_CN": "列名"
+		}
+	}]
+}
diff --git a/etc/sinks/redis.json b/extensions/sinks/redis/redis.json
similarity index 96%
rename from etc/sinks/redis.json
rename to extensions/sinks/redis/redis.json
index 72b6b820..f115d146 100644
--- a/etc/sinks/redis.json
+++ b/extensions/sinks/redis/redis.json
@@ -129,5 +129,13 @@
 				"zh_CN": "超时时间"
 			}
 		}
-	]
+	],
+	"node": {
+		"category": "Sink",
+		"icon": "iconPath",
+		"label": {
+			"en": "Redis",
+			"zh": "Redis"
+		}
+	}
 }
diff --git a/etc/sinks/sql.json b/extensions/sinks/sql/sql.json
similarity index 94%
rename from etc/sinks/sql.json
rename to extensions/sinks/sql/sql.json
index 00bdbca2..78acb756 100644
--- a/etc/sinks/sql.json
+++ b/extensions/sinks/sql/sql.json
@@ -75,5 +75,13 @@
 			"zh_CN": "表字段"
 		}
 	}
- ]
+ ],
+	"node": {
+		"category": "Sink",
+		"icon": "iconPath",
+		"label": {
+			"en": "SQL",
+			"zh": "SQL"
+		}
+	}
 }
diff --git a/etc/sinks/tdengine.json b/extensions/sinks/tdengine/tdengine.json
similarity index 97%
rename from etc/sinks/tdengine.json
rename to extensions/sinks/tdengine/tdengine.json
index a42e668c..61408c56 100644
--- a/etc/sinks/tdengine.json
+++ b/extensions/sinks/tdengine/tdengine.json
@@ -199,5 +199,13 @@
         "zh_CN": "标签字段"
       }
     }
-  ]
+  ],
+  "node": {
+    "category": "Sink",
+    "icon": "iconPath",
+    "label": {
+      "en": "Tdengine",
+      "zh": "Tdengine"
+    }
+  }
 }
diff --git a/etc/sinks/zmq.json b/extensions/sinks/zmq/zmq.json
similarity index 91%
rename from etc/sinks/zmq.json
rename to extensions/sinks/zmq/zmq.json
index 11897e23..1b3e36e7 100644
--- a/etc/sinks/zmq.json
+++ b/extensions/sinks/zmq/zmq.json
@@ -47,5 +47,13 @@
 			"en_US": "Topic",
 			"zh_CN": "主题"
 		}
-	}]
+	}],
+	"node": {
+		"category": "Sink",
+		"icon": "iconPath",
+		"label": {
+			"en": "ZMQ",
+			"zh": "ZMQ"
+		}
+	}
 }
diff --git a/etc/sources/random.json b/extensions/sources/random/random.json
similarity index 92%
rename from etc/sources/random.json
rename to extensions/sources/random/random.json
index e0c9be03..d71e2156 100644
--- a/etc/sources/random.json
+++ b/extensions/sources/random/random.json
@@ -96,5 +96,22 @@
         }
       }
     ]
+  },
+  "outputs": [
+    {
+      "label": {
+        "en_US": "Output",
+        "zh_CN": "输出"
+      },
+      "value": "signal"
+    }
+  ],
+  "node": {
+    "category": ["Protocol","Protocol1"],
+    "icon": "iconPath",
+    "label": {
+      "en_US": "Random",
+      "zh_CN": "Random"
+    }
   }
 }
diff --git a/etc/sources/random.yaml b/extensions/sources/random/random.yaml
similarity index 100%
rename from etc/sources/random.yaml
rename to extensions/sources/random/random.yaml
diff --git a/etc/sources/sql.json b/extensions/sources/sql/sql.json
similarity index 96%
rename from etc/sources/sql.json
rename to extensions/sources/sql/sql.json
index 9c5b2e73..89347e04 100644
--- a/etc/sources/sql.json
+++ b/extensions/sources/sql/sql.json
@@ -249,5 +249,22 @@
         }
       }
     ]
+  },
+  "outputs": [
+    {
+      "label": {
+        "en_US": "Output",
+        "zh_CN": "输出"
+      },
+      "value": "signal"
+    }
+  ],
+  "node": {
+    "category": ["Protocol","Protocol1"],
+    "icon": "iconPath",
+    "label": {
+      "en_US": "SQL",
+      "zh_CN": "SQL"
+    }
   }
 }
diff --git a/etc/sources/sql.yaml b/extensions/sources/sql/sql.yaml
similarity index 100%
rename from etc/sources/sql.yaml
rename to extensions/sources/sql/sql.yaml
diff --git a/etc/sources/zmq.json b/extensions/sources/zmq/zmq.json
similarity index 83%
rename from etc/sources/zmq.json
rename to extensions/sources/zmq/zmq.json
index 2b3038b6..ec976c96 100644
--- a/etc/sources/zmq.json
+++ b/extensions/sources/zmq/zmq.json
@@ -52,5 +52,22 @@
         }
       }
     ]
+  },
+  "outputs": [
+    {
+      "label": {
+        "en_US": "Output",
+        "zh_CN": "输出"
+      },
+      "value": "signal"
+    }
+  ],
+  "node": {
+    "category": ["Protocol","Protocol1"],
+    "icon": "iconPath",
+    "label": {
+      "en_US": "ZMQ",
+      "zh_CN": "ZMQ"
+    }
   }
 }
diff --git a/etc/sources/zmq.yaml b/extensions/sources/zmq/zmq.yaml
similarity index 100%
rename from etc/sources/zmq.yaml
rename to extensions/sources/zmq/zmq.yaml
diff --git a/internal/meta/func_meta.go b/internal/meta/func_meta.go
index 1027683a..acbeeaa4 100644
--- a/internal/meta/func_meta.go
+++ b/internal/meta/func_meta.go
@@ -25,23 +25,33 @@ import (
 
 type (
 	fileFunc struct {
-		Name    string        `json:"name"`
-		Example string        `json:"example"`
-		Hint    *fileLanguage `json:"hint"`
+		Name       string        `json:"name"`
+		Example    string        `json:"example"`
+		Hint       *fileLanguage `json:"hint"`
+		Aggregate  bool          `json:"aggregate"`
+		ArgsFields []*fileField  `json:"args"`
+		Node       *fileNode     `json:"node"`
+		Outputs    []interface{} `json:"outputs"`
 	}
 	fileFuncs struct {
 		About   *fileAbout  `json:"about"`
 		Name    string      `json:"name"`
+		Version string      `json:"version"`
 		FiFuncs []*fileFunc `json:"functions"`
 	}
 	uiFunc struct {
-		Name    string    `json:"name"`
-		Example string    `json:"example"`
-		Hint    *language `json:"hint"`
+		Name       string        `json:"name"`
+		Example    string        `json:"example"`
+		Hint       *language     `json:"hint"`
+		Aggregate  bool          `json:"aggregate"`
+		ArgsFields []*fileField  `json:"args"`
+		Node       *node         `json:"node"`
+		Outputs    []interface{} `json:"outputs"`
 	}
 	uiFuncs struct {
 		About   *about    `json:"about"`
 		Name    string    `json:"name"`
+		Version string    `json:"version"`
 		UiFuncs []*uiFunc `json:"functions"`
 	}
 )
@@ -58,6 +68,13 @@ func newUiFuncs(fi *fileFuncs) *uiFuncs {
 		ui.Name = v.Name
 		ui.Example = v.Example
 		ui.Hint = newLanguage(v.Hint)
+		ui.Aggregate = v.Aggregate
+		ui.ArgsFields = v.ArgsFields
+		ui.Node = newNode(v.Node)
+		ui.Outputs = make([]interface{}, len(v.Outputs))
+		for k, field := range v.Outputs {
+			ui.Outputs[k] = field
+		}
 		uis.UiFuncs = append(uis.UiFuncs, ui)
 	}
 	return uis
diff --git a/internal/meta/sinkMeta.go b/internal/meta/sinkMeta.go
index 53a5b167..efcbd013 100644
--- a/internal/meta/sinkMeta.go
+++ b/internal/meta/sinkMeta.go
@@ -58,10 +58,16 @@ type (
 		HelpUrl     *fileLanguage `json:"helpUrl"`
 		Description *fileLanguage `json:"description"`
 	}
+	fileNode struct {
+		Category string        `json:"category"`
+		Icon     string        `json:"iconPath"`
+		Label    *fileLanguage `json:"label"`
+	}
 	fileSink struct {
 		About  *fileAbout   `json:"about"`
 		Libs   []string     `json:"libs"`
 		Fields []*fileField `json:"properties"`
+		Node   *fileNode    `json:"node"`
 	}
 	language struct {
 		English string `json:"en"`
@@ -86,11 +92,16 @@ type (
 		Hint              *language   `json:"hint"`
 		Label             *language   `json:"label"`
 	}
-
+	node struct {
+		Category string    `json:"category"`
+		Icon     string    `json:"iconPath"`
+		Label    *language `json:"label"`
+	}
 	uiSink struct {
 		About  *about   `json:"about"`
 		Libs   []string `json:"libs"`
 		Fields []field  `json:"properties"`
+		Node   *node    `json:"node"`
 	}
 )
 
@@ -146,6 +157,16 @@ func newAbout(fi *fileAbout) *about {
 	ui.Description = newLanguage(fi.Description)
 	return ui
 }
+func newNode(fi *fileNode) *node {
+	if nil == fi {
+		return nil
+	}
+	ui := new(node)
+	ui.Category = fi.Category
+	ui.Icon = fi.Icon
+	ui.Label = newLanguage(fi.Label)
+	return ui
+}
 func newUiSink(fi *fileSink) (*uiSink, error) {
 	if nil == fi {
 		return nil, nil
@@ -154,6 +175,7 @@ func newUiSink(fi *fileSink) (*uiSink, error) {
 	ui := new(uiSink)
 	ui.Libs = fi.Libs
 	ui.About = newAbout(fi.About)
+	ui.Node = newNode(fi.Node)
 	ui.Fields, err = newField(fi.Fields)
 	return ui, err
 }
diff --git a/internal/meta/sourceMeta.go b/internal/meta/sourceMeta.go
index f7711926..f7c21316 100644
--- a/internal/meta/sourceMeta.go
+++ b/internal/meta/sourceMeta.go
@@ -29,11 +29,15 @@ type (
 		About    *fileAbout              `json:"about"`
 		Libs     []string                `json:"libs"`
 		ConfKeys map[string][]*fileField `json:"properties"`
+		Node     *fileNode               `json:"node"`
+		Outputs  []interface{}           `json:"outputs"`
 	}
 	uiSource struct {
 		About    *about             `json:"about"`
 		Libs     []string           `json:"libs"`
 		ConfKeys map[string][]field `json:"properties"`
+		Node     *node              `json:"node"`
+		Outputs  []interface{}      `json:"outputs"`
 	}
 )
 
@@ -45,8 +49,12 @@ func newUiSource(fi *fileSource) (*uiSource, error) {
 	ui := new(uiSource)
 	ui.Libs = fi.Libs
 	ui.About = newAbout(fi.About)
+	ui.Node = newNode(fi.Node)
+	ui.Outputs = make([]interface{}, len(fi.Outputs))
+	for k, field := range fi.Outputs {
+		ui.Outputs[k] = field
+	}
 	ui.ConfKeys = make(map[string][]field)
-
 	for k, fields := range fi.ConfKeys {
 		if ui.ConfKeys[k], err = newField(fields); nil != err {
 			return nil, err
diff --git a/internal/plugin/native/manager.go b/internal/plugin/native/manager.go
index a98067d2..f8969ce5 100644
--- a/internal/plugin/native/manager.go
+++ b/internal/plugin/native/manager.go
@@ -340,10 +340,16 @@ func (rr *Manager) Delete(t plugin2.PluginType, name string, stop bool) error {
 	switch t {
 	case plugin2.SOURCE:
 		paths = append(paths, path.Join(rr.etcDir, plugin2.PluginTypes[t], name+".yaml"))
+		srcJsonPath := path.Join(rr.etcDir, plugin2.PluginTypes[plugin2.SOURCE], name+".json")
+		_ = os.RemoveAll(srcJsonPath)
 		meta.UninstallSource(name)
 	case plugin2.SINK:
+		sinkJsonPaths := path.Join(rr.etcDir, plugin2.PluginTypes[plugin2.SINK], name+".json")
+		_ = os.RemoveAll(sinkJsonPaths)
 		meta.UninstallSink(name)
 	case plugin2.FUNCTION:
+		funcJsonPath := path.Join(rr.etcDir, plugin2.PluginTypes[plugin2.FUNCTION], name+".json")
+		_ = os.RemoveAll(funcJsonPath)
 		old := make([]string, 0)
 		if ok, err := rr.db.Get(name, &old); err != nil {
 			return err
diff --git a/internal/topo/node/sink_node.go b/internal/topo/node/sink_node.go
index 260d514d..26803e07 100644
--- a/internal/topo/node/sink_node.go
+++ b/internal/topo/node/sink_node.go
@@ -280,8 +280,8 @@ func (m *SinkNode) parseConf(logger api.Logger) (*SinkConf, error) {
 	m.concurrency = sconf.Concurrency
 	if sconf.Format == "" {
 		sconf.Format = "json"
-	} else if sconf.Format != message.FormatJson && sconf.Format != message.FormatProtobuf {
-		logger.Warnf("invalid type for format property, should be json or protobuf but found %s", sconf.Format)
+	} else if sconf.Format != message.FormatJson && sconf.Format != message.FormatProtobuf && sconf.Format != message.FormatBinary {
+		logger.Warnf("invalid type for format property, should be json protobuf or binary but found %s", sconf.Format)
 		sconf.Format = "json"
 	}
 	err = cast.MapToStruct(m.options, &sconf.SinkConf)


Test patch:
diff --git a/.github/workflows/run_test_case.yaml b/.github/workflows/run_test_case.yaml
index 63e75738..fdb7015c 100644
--- a/.github/workflows/run_test_case.yaml
+++ b/.github/workflows/run_test_case.yaml
@@ -43,6 +43,7 @@ jobs:
         mkdir -p log
         sed -i -r "s/debug: .*/debug: true/1" etc/kuiper.yaml
         go build -modfile extensions.mod --buildmode=plugin -o plugins/sources/Random@v2.0.0.so extensions/sources/random/random.go
+        cp extensions/sources/random/random.yaml etc/sources/random.yaml
         go build -modfile extensions.mod --buildmode=plugin -o plugins/sinks/File@v1.0.0.so extensions/sinks/file/file.go
         go build -modfile extensions.mod --buildmode=plugin -o plugins/functions/Echo.so extensions/functions/echo/echo.go
         go build -modfile extensions.mod --buildmode=plugin -o plugins/functions/CountPlusOne@v1.0.0.so extensions/functions/countPlusOne/countPlusOne.go
diff --git a/etc/sources/test.json b/etc/sources/test.json
deleted file mode 100644
index 7239ac7e..00000000
--- a/etc/sources/test.json
+++ /dev/null
@@ -1,66 +0,0 @@
-{
-  "libs": [],
-  "about": {
-    "trial": false,
-    "author": {
-      "name": "EMQ",
-      "email": "contact@emqx.io",
-      "company": "EMQ Technologies Co., Ltd",
-      "website": "https://www.emqx.io"
-    },
-    "helpUrl": {
-      "en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/rules/sources/builtin/test.md",
-      "zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/rules/sources/builtin/test.md"
-    },
-    "description": {
-      "en_US": "Desc US",
-      "zh_CN": "Desc CH"
-    }
-  },
-  "properties": {
-    "default": [
-      {
-        "name": "FIELDS",
-        "optional": false,
-        "control": "text",
-        "type": "array",
-        "hint": {
-          "en_US": "test",
-          "zh_CN": "test"
-        },
-        "label": {
-          "en_US": "FIELDS",
-          "zh_CN": "FIELDS"
-        }
-      },
-      {
-        "name": "QUERY_TABLE",
-        "optional": false,
-        "control": "text",
-        "type": "string",
-        "hint": {
-          "en_US": "test",
-          "zh_CN": "test"
-        },
-        "label": {
-          "en_US": "QUERY_TABLE",
-          "zh_CN": "QUERY_TABLE"
-        }
-      },
-      {
-        "name": "ROWCOUNT",
-        "optional": false,
-        "control": "number",
-        "type": "int",
-        "hint": {
-          "en_US": "test",
-          "zh_CN": "test"
-        },
-        "label": {
-          "en_US": "ROWCOUNT",
-          "zh_CN": "ROWCOUNT"
-        }
-      }
-    ]
-  }
-}
\ No newline at end of file
diff --git a/etc/sources/test.yaml b/etc/sources/test.yaml
deleted file mode 100644
index f1c14617..00000000
--- a/etc/sources/test.yaml
+++ /dev/null
@@ -1,15 +0,0 @@
-default:
-  interval: 1000
-  ashost: 192.168.1.100
-  sysnr: "02"
-  client: "900"
-  user: SPERF
-  passwd: PASSPASS
-  params:
-    QUERY_TABLE: VBAP
-    ROWCOUNT: 10
-    FIELDS: [
-      FIELDNAME: MANDT,
-      FIELDNAME: VBELN,
-      FIELDNAME: POSNR,
-    ]
\ No newline at end of file
diff --git a/extensions/sinks/influx2/influx2_test.go b/extensions/sinks/influx2/influx2_test.go
new file mode 100644
index 00000000..c4700a78
--- /dev/null
+++ b/extensions/sinks/influx2/influx2_test.go
@@ -0,0 +1,119 @@
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// 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.
+
+package main
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/testx"
+	"reflect"
+	"testing"
+)
+
+func TestConfig(t *testing.T) {
+	var tests = []struct {
+		conf     map[string]interface{}
+		expected *influxSink2
+		error    string
+	}{
+		{ //0
+			conf: map[string]interface{}{
+				"addr":        "http://192.168.0.3:8086",
+				"token":       "Token_test",
+				"measurement": "test",
+				"org":         "admin",
+				"bucket":      "bucket_one",
+				"tagKey":      "tag",
+				"tagValue":    "value",
+				"fields":      "temperature",
+			},
+			expected: &influxSink2{
+				addr:         "http://192.168.0.3:8086",
+				token:        "Token_test",
+				measurement:  "test",
+				org:          "admin",
+				bucket:       "bucket_one",
+				tagKey:       "tag",
+				tagValue:     "value",
+				fields:       "temperature",
+				cli:          nil,
+				fieldMap:     nil,
+				hasTransform: false,
+			},
+		},
+		{ //1
+			conf: map[string]interface{}{
+				"addr":         "http://192.168.0.3:8086",
+				"token":        "Token_test",
+				"measurement":  "test",
+				"org":          "admin",
+				"bucket":       "bucket_one",
+				"tagKey":       "tag",
+				"tagValue":     "value",
+				"fields":       "temperature",
+				"dataTemplate": "",
+			},
+			expected: &influxSink2{
+				addr:         "http://192.168.0.3:8086",
+				token:        "Token_test",
+				measurement:  "test",
+				org:          "admin",
+				bucket:       "bucket_one",
+				tagKey:       "tag",
+				tagValue:     "value",
+				fields:       "temperature",
+				cli:          nil,
+				fieldMap:     nil,
+				hasTransform: false,
+			},
+		},
+		{ //2
+			conf: map[string]interface{}{
+				"addr":         "http://192.168.0.3:8086",
+				"token":        "Token_test",
+				"measurement":  "test",
+				"org":          "admin",
+				"bucket":       "bucket_one",
+				"tagKey":       "tag",
+				"tagValue":     "value",
+				"fields":       "temperature",
+				"dataTemplate": "{{funcname .arg1 .arg2}}",
+			},
+			expected: &influxSink2{
+				addr:         "http://192.168.0.3:8086",
+				token:        "Token_test",
+				measurement:  "test",
+				org:          "admin",
+				bucket:       "bucket_one",
+				tagKey:       "tag",
+				tagValue:     "value",
+				fields:       "temperature",
+				cli:          nil,
+				fieldMap:     nil,
+				hasTransform: true,
+			},
+		},
+	}
+
+	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
+	for i, test := range tests {
+		ifsink := &influxSink2{}
+		err := ifsink.Configure(test.conf)
+		if !reflect.DeepEqual(test.error, testx.Errstring(err)) {
+			t.Errorf("%d: error mismatch:\n  exp=%s\n  got=%s\n\n", i, test.error, err)
+		} else if test.error == "" && !reflect.DeepEqual(test.expected, ifsink) {
+			t.Errorf("%d\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, test.expected, ifsink)
+		}
+	}
+}
diff --git a/internal/meta/sourceMeta_test.go b/internal/meta/sourceMeta_test.go
index 6c452593..4e571544 100644
--- a/internal/meta/sourceMeta_test.go
+++ b/internal/meta/sourceMeta_test.go
@@ -51,12 +51,12 @@ func TestGetSqlSourceMeta(t *testing.T) {
 		return
 	}
 
-	if err = ReadSourceMetaFile(path.Join(confDir, "sources", "sql.json"), true); nil != err {
+	if err = ReadSourceMetaFile(path.Join(confDir, "sources", "httppull.json"), true); nil != err {
 		t.Error(err)
 		return
 	}
 
-	showMeta, err := GetSourceMeta("sql", "zh_CN")
+	showMeta, err := GetSourceMeta("httppull", "zh_CN")
 	if nil != err {
 		t.Error(err)
 	}
@@ -76,12 +76,12 @@ func TestGetSqlSinkMeta(t *testing.T) {
 		return
 	}
 
-	if err = ReadSinkMetaFile(path.Join(confDir, "sinks", "sql.json"), true); nil != err {
+	if err = ReadSinkMetaFile(path.Join(confDir, "sinks", "mqtt.json"), true); nil != err {
 		t.Error(err)
 		return
 	}
 
-	showMeta, err := GetSinkMeta("sql", "zh_CN")
+	showMeta, err := GetSinkMeta("mqtt", "zh_CN")
 	if nil != err {
 		t.Error(err)
 		return
diff --git a/internal/topo/node/source_node_test.go b/internal/topo/node/source_node_test.go
index 96cd5ad3..2322dda4 100644
--- a/internal/topo/node/source_node_test.go
+++ b/internal/topo/node/source_node_test.go
@@ -25,26 +25,22 @@ import (
 
 func TestGetConf_Apply(t *testing.T) {
 	result := map[string]interface{}{
-		"interval": 1000,
-		"ashost":   "192.168.1.100",
-		"sysnr":    "02",
-		"client":   "900",
-		"user":     "SPERF",
-		"passwd":   "PASSPASS",
-		"format":   "json",
-		"params": map[string]interface{}{
-			"QUERY_TABLE": "VBAP",
-			"ROWCOUNT":    10,
-			"FIELDS": []interface{}{
-				map[string]interface{}{"FIELDNAME": "MANDT"},
-				map[string]interface{}{"FIELDNAME": "VBELN"},
-				map[string]interface{}{"FIELDNAME": "POSNR"},
-			},
+		"url":                "http://localhost",
+		"method":             "post",
+		"interval":           10000,
+		"timeout":            5000,
+		"incremental":        false,
+		"body":               "{}",
+		"bodyType":           "json",
+		"format":             "json",
+		"insecureSkipVerify": true,
+		"headers": map[string]interface{}{
+			"Accept": "application/json",
 		},
 	}
 	n := NewSourceNode("test", ast.TypeStream, nil, &ast.Options{
-		DATASOURCE: "RFC_READ_TABLE",
-		TYPE:       "test",
+		DATASOURCE: "/feed",
+		TYPE:       "httppull",
 	}, false)
 	contextLogger := conf.Log.WithField("rule", "test")
 	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
@@ -56,18 +52,23 @@ func TestGetConf_Apply(t *testing.T) {
 
 func TestGetConfAndConvert_Apply(t *testing.T) {
 	result := map[string]interface{}{
-		"interval": 100,
-		"seed":     1,
-		"format":   "json",
-		"pattern": map[string]interface{}{
-			"count": 50,
+		"url":                "http://localhost:9090/",
+		"method":             "post",
+		"interval":           10000,
+		"timeout":            5000,
+		"incremental":        true,
+		"body":               "{}",
+		"bodyType":           "json",
+		"format":             "json",
+		"insecureSkipVerify": true,
+		"headers": map[string]interface{}{
+			"Accept": "application/json",
 		},
-		"deduplicate": 50,
 	}
 	n := NewSourceNode("test", ast.TypeStream, nil, &ast.Options{
-		DATASOURCE: "test",
-		TYPE:       "random",
-		CONF_KEY:   "dedup",
+		DATASOURCE: "/feed",
+		TYPE:       "httppull",
+		CONF_KEY:   "application_conf",
 	}, false)
 	contextLogger := conf.Log.WithField("rule", "test")
 	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
@@ -77,16 +78,21 @@ func TestGetConfAndConvert_Apply(t *testing.T) {
 		return
 	}
 
-	r := &randomSourceConfig{
-		Interval: 100,
-		Seed:     1,
-		Pattern: map[string]interface{}{
-			"count": 50,
+	r := &httpPullSourceConfig{
+		Url:                "http://localhost:9090/",
+		Method:             "post",
+		Interval:           10000,
+		Timeout:            5000,
+		Incremental:        true,
+		Body:               "{}",
+		BodyType:           "json",
+		InsecureSkipVerify: true,
+		Headers: map[string]interface{}{
+			"Accept": "application/json",
 		},
-		Deduplicate: 50,
 	}
 
-	cfg := &randomSourceConfig{}
+	cfg := &httpPullSourceConfig{}
 	err := cast.MapToStruct(conf, cfg)
 	if err != nil {
 		t.Errorf("map to sturct error %s", err)
@@ -99,9 +105,14 @@ func TestGetConfAndConvert_Apply(t *testing.T) {
 	}
 }
 
-type randomSourceConfig struct {
-	Interval    int                    `json:"interval"`
-	Seed        int                    `json:"seed"`
-	Pattern     map[string]interface{} `json:"pattern"`
-	Deduplicate int                    `json:"deduplicate"`
+type httpPullSourceConfig struct {
+	Url                string                 `json:"url"`
+	Method             string                 `json:"method"`
+	Interval           int                    `json:"interval"`
+	Timeout            int                    `json:"timeout"`
+	Incremental        bool                   `json:"incremental"`
+	Body               string                 `json:"body"`
+	BodyType           string                 `json:"bodyType"`
+	InsecureSkipVerify bool                   `json:"insecureSkipVerify"`
+	Headers            map[string]interface{} `json:"headers"`
 }