add plugin-sink with influxDB (#311)

* add plugin-sink with influxDB

* add plugin-sink with influxDB

* Modify file influxdb.go,overview.md
Add file influxdb.md

* Modify file influxdb.md,influxdb.go

* Modify file influxdb.md,influxdb.go

* use gofmt to format code

* use gofmt to format code

* use gofmt to format code

Co-authored-by: 726518972@qq.com <Smart33690>

docs/en_US/plugins/overview.md

@@ -15,7 +15,7 @@ Kuiper implemented several plugins.
 | --------------------- | ------------------------------------------------------------ |
 | [file](sinks/file.md) | The sink is used for saving analysis result into a specified file. |
 | [zmq](sinks/zmq.md)   | The sink will publish the result into a Zero Mq topic                                                             |
-
+| [influxdb](sinks/influxdb.md)   | The sink will publish the result into a InfluxDB 
 
 
 

docs/en_US/plugins/sinks/influxdb.md

@@ -0,0 +1,74 @@
+# InfluxDB Sink
+
+The sink will publish the result into a InfluxDB.
+
+## Compile & deploy plugin
+
+```shell
+# cd $kuiper_src
+# go build --buildmode=plugin -o plugins/sinks/InfluxDB.so plugins/sinks/influxdb.go
+# zip influx.zip plugins/sinks/InfluxDB.so
+# cp influx.zip /root/tomcat_path/webapps/ROOT/
+# bin/cli create plugin sink influx -f /tmp/influxPlugin.txt
+# bin/cli create rule influx -f /tmp/influxRule.txt
+```
+
+Restart the Kuiper 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) |
+| username      | false    | The InfluxDB login username |
+| password      | false    | The InfluxDB login password |
+| databasename  | true     | The database of the InfluxDB |
+| tagkey        | true     | The tag key of the InfluxDB |
+| tagvalue      | true     | The tag value of the InfluxDB |
+## 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": {},
+      "influx":{
+       "addr": "http://192.168.100.245:8086",
+       "username": "",
+       "password": "",
+       "measurement": "test",
+       "databasename": "databasename",
+       "tagkey": "tagkey",
+       "tagvalue": "tagvalue"
+      }
+    }
+  ]
+}
+```
+####/tmp/influxPlugin.txt
+```json
+{
+   "file":"http://localhost:8080/influx.zip"
+ }
+```
+#### plugins/go.mod
+```
+module plugins
+
+go 1.14
+
+require (
+        github.com/emqx/kuiper v0.0.0-20200323140757-60d00241372b
+        github.com/influxdata/influxdb-client-go v1.2.0
+        github.com/influxdata/influxdb1-client v0.0.0-20200515024757-02f0bf5dbca3 // indirect
+)
+
+replace github.com/emqx/kuiper => /root/goProject/kuiper
+
+```

docs/zh_CN/plugins/overview.md

@@ -17,7 +17,7 @@ Kuiper 实现了几个插件。
 | --------------------- | ------------------------------------------------------------ |
 | [file](sinks/file.md) | 该插件将分析结果保存到某个指定到文件系统中 |
 | [zmq](sinks/zmq.md)   | 该插件将分析结果发送到Zero Mq的主题中    |
-
+| [influxdb](sinks/influxdb.md)   | 该插件将分析结果发送到InfluxDB中    |
 
 
 

docs/zh_CN/plugins/sinks/influxdb.md

@@ -0,0 +1,73 @@
+# InfluxDB Sink
+
+该插件将分析结果发送到InfluxDB中
+## 编译插件&创建插件
+
+```shell
+# cd $kuiper_src
+# go build --buildmode=plugin -o plugins/sinks/InfluxDB.so plugins/sinks/influxdb.go
+# zip influx.zip plugins/sinks/InfluxDB.so
+# cp influx.zip /root/tomcat_path/webapps/ROOT/
+# bin/cli create plugin sink influx -f /tmp/influxPlugin.txt
+# bin/cli create rule influx -f /tmp/influxRule.txt
+```
+
+重新启动Kuiper服务器以激活插件。
+
+## 属性
+
+| Property name | Optional | Description                                                  |
+| ------------- | -------- | ------------------------------------------------------------ |
+| addr          | true     | The addr of the InfluxDB |
+| measurement   | true     | The measurement of the InfluxDb (like table name) |
+| username      | false    | The InfluxDB login username |
+| password      | false    | The InfluxDB login password |
+| databasename  | true     | The database of the InfluxDB |
+| tagkey        | true     | The tag key of the InfluxDB |
+| tagvalue      | true     | The tag value of the InfluxDB |
+## 示例用法
+
+下面是选择温度大于50度的样本规则，和一些配置文件仅供参考。
+
+#### /tmp/influxRule.txt
+```json
+{
+  "id": "influx",
+  "sql": "SELECT * from  demo_stream where temperature < 50",
+  "actions": [
+    {
+      "log": {},
+      "influx":{
+       "addr": "http://192.168.100.245:8086",
+       "username": "",
+       "password": "",
+       "measurement": "test",
+       "databasename": "databasename",
+       "tagkey": "tagkey",
+       "tagvalue": "tagvalue"
+      }
+    }
+  ]
+}
+```
+#### /tmp/influxPlugin.txt
+```json
+{
+  "file":"http://localhost:8080/influx.zip"
+}
+```
+#### plugins/go.mod
+```
+module plugins
+
+go 1.14
+
+require (
+        github.com/emqx/kuiper v0.0.0-20200323140757-60d00241372b
+        github.com/influxdata/influxdb-client-go v1.2.0
+        github.com/influxdata/influxdb1-client v0.0.0-20200515024757-02f0bf5dbca3 // indirect
+)
+
+replace github.com/emqx/kuiper => /root/goProject/kuiper
+
+```

plugins/sinks/influxdb.go

@@ -0,0 +1,128 @@
+package main
+
+import (
+	"encoding/json"
+	api "github.com/emqx/kuiper/xstream/api"
+	_ "github.com/influxdata/influxdb1-client/v2"
+	client "github.com/influxdata/influxdb1-client/v2"
+	"time"
+)
+
+type influxSink struct {
+	addr         string
+	username     string
+	password     string
+	measurement  string
+	databasename string
+	tagkey       string
+	tagvalue     string
+}
+
+var cli client.Client
+
+type ListMap []map[string]float64
+
+func (m *influxSink) 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["username"]; ok {
+		if i, ok := i.(string); ok {
+			m.username = i
+		}
+	}
+	if i, ok := props["password"]; ok {
+		if i, ok := i.(string); ok {
+			m.password = i
+		}
+	}
+	if i, ok := props["measurement"]; ok {
+		if i, ok := i.(string); ok {
+			m.measurement = i
+		}
+	}
+	if i, ok := props["databasename"]; ok {
+		if i, ok := i.(string); ok {
+			m.databasename = 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
+		}
+	}
+	return nil
+}
+
+func (m *influxSink) Open(ctx api.StreamContext) (err error) {
+	logger := ctx.GetLogger()
+	logger.Debug("Opening influx sink")
+	cli, err = client.NewHTTPClient(client.HTTPConfig{
+		Addr:     m.addr,
+		Username: m.username,
+		Password: m.password,
+	})
+	if err != nil {
+		logger.Debug(err)
+		return err
+	}
+	return nil
+}
+
+func (m *influxSink) Collect(ctx api.StreamContext, data interface{}) error {
+	logger := ctx.GetLogger()
+	if v, ok := data.([]byte); ok {
+		var out ListMap
+		if err := json.Unmarshal([]byte(v), &out); err != nil {
+			logger.Debug("Failed to unmarshal data with error %s.\n", err)
+			return err
+		}
+		bp, err := client.NewBatchPoints(client.BatchPointsConfig{
+			Database:  m.databasename,
+			Precision: "ns", //default is ns
+		})
+		if err != nil {
+			logger.Debug(err)
+			return err
+		}
+		tags := map[string]string{m.tagkey: m.tagvalue}
+		fields := map[string]interface{}{
+			"temperature": out[0]["temperature"],
+			"humidity":    out[0]["humidity"],
+		}
+
+		pt, err := client.NewPoint(m.measurement, tags, fields, time.Now())
+		if err != nil {
+			logger.Debug(err)
+			return err
+		}
+		bp.AddPoint(pt)
+		err = cli.Write(bp)
+		if err != nil {
+			logger.Debug(err)
+			return err
+		}
+		logger.Debug("insert success")
+	} else {
+		logger.Debug("insert faild")
+	}
+	return nil
+}
+
+func (m *influxSink) Close(ctx api.StreamContext) error {
+	// Close the client
+	// Close the client
+	cli.Close()
+	return nil
+}
+
+func Influx() api.Sink {
+	return &influxSink{}
+}