feat(plugin): add support for the sink plug-in of the influx V2 version (#1396)

* feat(plugin): add support for the sink plug-in of the influx V2 version

Signed-off-by: crui <crui@relaper.com>

* feat(plugin): add influxV2 sink plug-in unit test

Signed-off-by: crui <crui@relaper.com>

* feat(plugin): add influxV2 doc

Signed-off-by: crui <crui@relaper.com>

* feat(plugin): add influxV2 path in directory.json

Signed-off-by: crui <crui@relaper.com>

* fix(plugin): modified some description information of influx_v2 doc.

Signed-off-by: crui <crui@relaper.com>

* fix(plugin): rename influx_v2 to influx2

Signed-off-by: crui <crui@relaper.com>

Signed-off-by: crui <crui@relaper.com>
Co-authored-by: crui <crui@relaper.com>

docs/directory.json

@@ -289,6 +289,10 @@
 									"path": "rules/sinks/plugin/influx"
 								},
 								{
+									"title": "InfluxDBV2 动作",
+									"path": "rules/sinks/plugin/influx2"
+								},
+								{
 									"title": "TDengine 动作",
 									"path": "rules/sinks/plugin/tdengine"
 								},
@@ -843,6 +847,10 @@
 									"path": "rules/sinks/plugin/influx"
 								},
 								{
+									"title": "InfluxDBV2 sink",
+									"path": "rules/sinks/plugin/influx2"
+								},
+								{
 									"title": "TDengine sink",
 									"path": "rules/sinks/plugin/tdengine"
 								},

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.

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
+
+```

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)： 写入一个图像文件。仅用于处理二进制结果。

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
+
+```

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": "列名"
+    }
+  }]
+}

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{}
+}

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": "列名"
+		}
+	}]
+}

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)
+		}
+	}
+}