Repository: lf-edge/ekuiper
Pull request #1293 (2022-06-08 02:51:57)

doc(neuron): update usage for alpine


Patch:
diff --git a/docs/en_US/rules/sinks/builtin/edgex.md b/docs/en_US/rules/sinks/builtin/edgex.md
index ac60acdb..b8284e04 100644
--- a/docs/en_US/rules/sinks/builtin/edgex.md
+++ b/docs/en_US/rules/sinks/builtin/edgex.md
@@ -39,6 +39,8 @@ Below optional configurations are supported, please check MQTT specification for
   - KeyPEMBlock
   - SkipCertVerify
 
+Notice that, the edgex action can support data templates to vary the result format, but the result of the data template must be in the object form of a JSON string, e.g. `"{\"key\":\"{{.key}}\"}"`. JSON strings in the form of arrays or non-JSON strings are not supported.
+
 ## Send to various targets
 
 By setting the combination of the properties, we can send the result to various EdgeX message bus settings.
@@ -270,5 +272,4 @@ Please notice that,
 - For the reading that can NOT be found in original message,  the metadata will not be set.  Such as metadata of `t1` in the sample will fill with default value that generated by eKuiper. 
 - If your SQL has aggregated function, then it does not make sense to keep these metadata, but eKuiper will still fill with metadata from a particular message in the time window. For example, with following SQL, 
 ```SELECT avg(temperature) AS temperature, meta(*) AS edgex_meta FROM ... GROUP BY TUMBLINGWINDOW(ss, 10)```. 
-In this case, there are possibly several messages in the window, the metadata value for `temperature` will be filled with value from 1st message that received from bus.
-
+In this case, there are possibly several messages in the window, the metadata value for `temperature` will be filled with value from 1st message that received from bus.
\ No newline at end of file
diff --git a/docs/en_US/tutorials/neuron/docker-compose.yml b/docs/en_US/tutorials/neuron/docker-compose.yml
index a772c901..d2d1e1f6 100644
--- a/docs/en_US/tutorials/neuron/docker-compose.yml
+++ b/docs/en_US/tutorials/neuron/docker-compose.yml
@@ -2,12 +2,12 @@ version: '3.4'
 
 services:
   manager:
-    image: emqx/ekuiper-manager:1.5.0
+    image: emqx/ekuiper-manager:1.5
     container_name: ekuiper-manager
     ports:
       - "9082:9082"
   ekuiper:
-    image: lfedge/ekuiper:1.5.0-slim
+    image: lfedge/ekuiper:1.5-slim
     ports:
       - "9081:9081"
       - "127.0.0.1:20498:20498"
diff --git a/docs/en_US/tutorials/neuron/neuron_integration_tutorial.md b/docs/en_US/tutorials/neuron/neuron_integration_tutorial.md
index 93f96250..2c101c9d 100644
--- a/docs/en_US/tutorials/neuron/neuron_integration_tutorial.md
+++ b/docs/en_US/tutorials/neuron/neuron_integration_tutorial.md
@@ -34,7 +34,7 @@ Before getting started, the following environment needs to be prepared.
 
 Both Neuron and eKuiper support binary installation packages and Docker containerized deployment solutions. In this article, we take Docker solution as an example and use [docker compose](https://docs.docker.com/compose/) to quickly deploy two components on the edge with one click.
 
-1. Copy [docker-compose.yml](. /docker-compose.yml) file to the deployment machine. The contents are as follows, containing Neuron, eKuiper, and the eKuiper manager (optional). In particular, eKuiper and neuron share a volume called nng-ipc, which is used for communication between the two.
+1. Copy [docker-compose.yml](. /docker-compose.yml) file to the deployment machine. The contents are as follows, containing Neuron, eKuiper, and the eKuiper manager (optional). In particular, eKuiper and neuron share a volume called nng-ipc, which is used for communication between the two. If using eKuiper alpine version, add `user: root:root` into the eKuiper section to grant the write access to the ipc file, otherwise, the connection cannot be established.
 
    ```yaml
    version: '3.4'
@@ -46,7 +46,7 @@ Both Neuron and eKuiper support binary installation packages and Docker containe
        ports:
          - "9082:9082"
      ekuiper:
-       image: lfedge/ekuiper:1.5.0-slim
+       image: lfedge/ekuiper:1.5-slim
        ports:
          - "9081:9081"
          - "127.0.0.1:20498:20498"
diff --git a/docs/zh_CN/rules/sinks/builtin/edgex.md b/docs/zh_CN/rules/sinks/builtin/edgex.md
index 9709b59b..7246fa0b 100644
--- a/docs/zh_CN/rules/sinks/builtin/edgex.md
+++ b/docs/zh_CN/rules/sinks/builtin/edgex.md
@@ -39,6 +39,8 @@
   - KeyPEMBlock
   - SkipCertVerify
 
+EdgeX 动作可支持数据模板对结果格式进行变化，但是数据模板的结果必须为 JSON 字符串的 object 形式，例如 `"{\"key\":\"{{.key}}\"}"`。数组形式的 JSON 字符串或者非 JSON 字符串都不支持。
+
 ## 发送到各种目标
 
 通过设置不同的属性组合，我们可以将结果采用不同的格式发送到不同的 EdgeX 消息总线设置中。
diff --git a/docs/zh_CN/tutorials/neuron/docker-compose.yml b/docs/zh_CN/tutorials/neuron/docker-compose.yml
index a772c901..d2d1e1f6 100644
--- a/docs/zh_CN/tutorials/neuron/docker-compose.yml
+++ b/docs/zh_CN/tutorials/neuron/docker-compose.yml
@@ -2,12 +2,12 @@ version: '3.4'
 
 services:
   manager:
-    image: emqx/ekuiper-manager:1.5.0
+    image: emqx/ekuiper-manager:1.5
     container_name: ekuiper-manager
     ports:
       - "9082:9082"
   ekuiper:
-    image: lfedge/ekuiper:1.5.0-slim
+    image: lfedge/ekuiper:1.5-slim
     ports:
       - "9081:9081"
       - "127.0.0.1:20498:20498"
diff --git a/docs/zh_CN/tutorials/neuron/neuron_integration_tutorial.md b/docs/zh_CN/tutorials/neuron/neuron_integration_tutorial.md
index ddef2498..83f18b5f 100644
--- a/docs/zh_CN/tutorials/neuron/neuron_integration_tutorial.md
+++ b/docs/zh_CN/tutorials/neuron/neuron_integration_tutorial.md
@@ -34,7 +34,7 @@ eKuiper 与 Neuron 之间的集成是双向的，其实现主要包含两个部
 
 Neuron 和 eKuiper 都支持二进制安装包以及 Docker 容器化部署方案。本文以 Docker 方案为例，采用 [docker compose](https://docs.docker.com/compose/) 方式，一键完成边缘端两个组件的快速部署。
 
-1. 复制 [docker-compose.yml](./docker-compose.yml) 文件到部署的机器上。其内容如下，包含了 Neuron，eKuiper 以及 eKuiper 的管理界面 eKuiper manager（可选）。其中，eKuiper 和 neuron 共享了名为 nng-ipc 的 volume ，用于二者通信。
+1. 复制 [docker-compose.yml](./docker-compose.yml) 文件到部署的机器上。其内容如下，包含了 Neuron，eKuiper 以及 eKuiper 的管理界面 eKuiper manager（可选）。其中，eKuiper 和 neuron 共享了名为 nng-ipc 的 volume ，用于二者通信。如果要使用 eKuiper alpine 版本，需要在 compose 文件的 eKuiper 部分添加 `user: root:root` 赋予写入 ipc 文件的权限，否则连接将无法建立。
 
    ```yaml
    version: '3.4'
@@ -46,7 +46,7 @@ Neuron 和 eKuiper 都支持二进制安装包以及 Docker 容器化部署方
        ports:
          - "9082:9082"
      ekuiper:
-       image: lfedge/ekuiper:1.5.0-slim
+       image: lfedge/ekuiper:1.5-slim
        ports:
          - "9081:9081"
          - "127.0.0.1:20498:20498"
diff --git a/extensions/sinks/influx/influx.go b/extensions/sinks/influx/influx.go
index ce032396..097899b8 100644
--- a/extensions/sinks/influx/influx.go
+++ b/extensions/sinks/influx/influx.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// 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.
@@ -30,9 +30,11 @@
 package main
 
 import (
+	"encoding/json"
+	"fmt"
 	_ "github.com/influxdata/influxdb1-client/v2"
 	client "github.com/influxdata/influxdb1-client/v2"
-	api "github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/api"
 	"strings"
 	"time"
 )
@@ -48,6 +50,7 @@ type influxSink struct {
 	fields       string
 	cli          client.Client
 	fieldmap     map[string]interface{}
+	hasTransform bool
 }
 
 func (m *influxSink) Configure(props map[string]interface{}) error {
@@ -91,6 +94,9 @@ func (m *influxSink) Configure(props map[string]interface{}) error {
 			m.fields = i
 		}
 	}
+	if _, ok := props["dataTemplate"]; ok {
+		m.hasTransform = true
+	}
 	return nil
 }
 
@@ -111,6 +117,18 @@ func (m *influxSink) Open(ctx api.StreamContext) (err error) {
 
 func (m *influxSink) 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{}:
diff --git a/extensions/sinks/sql/sql.go b/extensions/sinks/sql/sql.go
index f9d74ff0..5123d24a 100644
--- a/extensions/sinks/sql/sql.go
+++ b/extensions/sinks/sql/sql.go
@@ -16,6 +16,7 @@ package main
 
 import (
 	"database/sql"
+	"encoding/json"
 	"fmt"
 	"github.com/lf-edge/ekuiper/extensions/sqldatabase/driver"
 	"github.com/lf-edge/ekuiper/pkg/api"
@@ -26,9 +27,10 @@ import (
 )
 
 type sqlConfig struct {
-	Url    string   `json:"url"`
-	Table  string   `json:"table"`
-	Fields []string `json:"fields"`
+	Url          string   `json:"url"`
+	Table        string   `json:"table"`
+	Fields       []string `json:"fields"`
+	DataTemplate string   `json:"dataTemplate"`
 }
 
 func (t *sqlConfig) buildSql(ctx api.StreamContext, mapData map[string]interface{}) (string, error) {
@@ -126,7 +128,18 @@ func (m *sqlSink) writeToDB(ctx api.StreamContext, mapData map[string]interface{
 
 func (m *sqlSink) Collect(ctx api.StreamContext, item interface{}) error {
 	ctx.GetLogger().Debugf("sql sink receive %s", item)
-
+	if m.conf.DataTemplate != "" {
+		jsonBytes, _, err := ctx.TransformOutput(item)
+		if err != nil {
+			return err
+		}
+		tm := make(map[string]interface{})
+		err = json.Unmarshal(jsonBytes, &tm)
+		if err != nil {
+			return fmt.Errorf("fail to decode data %s after applying dataTemplate for error %v", string(jsonBytes), err)
+		}
+		item = tm
+	}
 	switch v := item.(type) {
 	case []map[string]interface{}:
 		var err error
diff --git a/extensions/sinks/tdengine/tdengine.go b/extensions/sinks/tdengine/tdengine.go
index dcfd9894..bd168e85 100644
--- a/extensions/sinks/tdengine/tdengine.go
+++ b/extensions/sinks/tdengine/tdengine.go
@@ -16,6 +16,7 @@ package main
 
 import (
 	"database/sql"
+	"encoding/json"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/pkg/api"
@@ -27,18 +28,19 @@ import (
 
 type (
 	taosConfig struct {
-		ProvideTs   bool     `json:"provideTs"`
-		Port        int      `json:"port"`
-		Ip          string   `json:"ip"` // To be deprecated
-		Host        string   `json:"host"`
-		User        string   `json:"user"`
-		Password    string   `json:"password"`
-		Database    string   `json:"database"`
-		Table       string   `json:"table"`
-		TsFieldName string   `json:"tsFieldName"`
-		Fields      []string `json:"fields"`
-		STable      string   `json:"sTable"`
-		TagFields   []string `json:"tagFields"`
+		ProvideTs    bool     `json:"provideTs"`
+		Port         int      `json:"port"`
+		Ip           string   `json:"ip"` // To be deprecated
+		Host         string   `json:"host"`
+		User         string   `json:"user"`
+		Password     string   `json:"password"`
+		Database     string   `json:"database"`
+		Table        string   `json:"table"`
+		TsFieldName  string   `json:"tsFieldName"`
+		Fields       []string `json:"fields"`
+		STable       string   `json:"sTable"`
+		TagFields    []string `json:"tagFields"`
+		DataTemplate string   `json:"dataTemplate"`
 	}
 	taosSink struct {
 		conf *taosConfig
@@ -191,7 +193,18 @@ func (m *taosSink) Open(ctx api.StreamContext) (err error) {
 
 func (m *taosSink) Collect(ctx api.StreamContext, item interface{}) error {
 	ctx.GetLogger().Debugf("tdengine sink receive %s", item)
-
+	if m.conf.DataTemplate != "" {
+		jsonBytes, _, err := ctx.TransformOutput(item)
+		if err != nil {
+			return err
+		}
+		tm := make(map[string]interface{})
+		err = json.Unmarshal(jsonBytes, &tm)
+		if err != nil {
+			return fmt.Errorf("fail to decode data %s after applying dataTemplate for error %v", string(jsonBytes), err)
+		}
+		item = tm
+	}
 	switch v := item.(type) {
 	case []map[string]interface{}:
 		var err error
diff --git a/internal/topo/sink/edgex_sink.go b/internal/topo/sink/edgex_sink.go
index 922729b7..468d1a0c 100644
--- a/internal/topo/sink/edgex_sink.go
+++ b/internal/topo/sink/edgex_sink.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// 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.
@@ -84,9 +84,6 @@ func (ems *EdgexMsgBusSink) Configure(ps map[string]interface{}) error {
 	if c.Topic != "" && c.TopicPrefix != "" {
 		return fmt.Errorf("not allow to specify both topic and topicPrefix, please set one only")
 	}
-	if c.DataTemplate != "" {
-		conf.Log.Warn("edgex sink does not support dataTemplate, just ignore")
-	}
 	ems.c = c
 	ems.config = ps
 
@@ -120,7 +117,29 @@ func (ems *EdgexMsgBusSink) Open(ctx api.StreamContext) error {
 	return nil
 }
 
-func (ems *EdgexMsgBusSink) produceEvents(ctx api.StreamContext, m []map[string]interface{}) (*dtos.Event, error) {
+func (ems *EdgexMsgBusSink) produceEvents(ctx api.StreamContext, item interface{}) (*dtos.Event, error) {
+	if ems.c.DataTemplate != "" {
+		jsonBytes, _, err := ctx.TransformOutput(item)
+		if err != nil {
+			return nil, err
+		}
+		tm := make(map[string]interface{})
+		err = json.Unmarshal(jsonBytes, &tm)
+		if err != nil {
+			return nil, fmt.Errorf("fail to decode data %s after applying dataTemplate for error %v", string(jsonBytes), err)
+		}
+		item = tm
+	}
+	var m []map[string]interface{}
+	switch payload := item.(type) {
+	case map[string]interface{}:
+		m = []map[string]interface{}{payload}
+	case []map[string]interface{}:
+		m = payload
+	default:
+		// impossible
+		return nil, fmt.Errorf("receive invalid data %v", item)
+	}
 	m1 := ems.getMeta(m)
 	event := m1.createEvent()
 	//Override the devicename if user specified the value
@@ -460,19 +479,7 @@ func (ems *EdgexMsgBusSink) getMeta(result []map[string]interface{}) *meta {
 
 func (ems *EdgexMsgBusSink) Collect(ctx api.StreamContext, item interface{}) error {
 	logger := ctx.GetLogger()
-	var (
-		evt *dtos.Event
-		err error
-	)
-	switch payload := item.(type) {
-	case map[string]interface{}:
-		evt, err = ems.produceEvents(ctx, []map[string]interface{}{payload})
-	case []map[string]interface{}:
-		evt, err = ems.produceEvents(ctx, payload)
-	default:
-		// impossible
-		return fmt.Errorf("receive invalid data %v", item)
-	}
+	evt, err := ems.produceEvents(ctx, item)
 	if err != nil {
 		return fmt.Errorf("Failed to convert to EdgeX event: %s.", err.Error())
 	}


Test patch:
diff --git a/internal/topo/sink/edgex_sink_test.go b/internal/topo/sink/edgex_sink_test.go
index f4a7a30a..4f5cc6e0 100644
--- a/internal/topo/sink/edgex_sink_test.go
+++ b/internal/topo/sink/edgex_sink_test.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// 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.
@@ -25,6 +25,8 @@ import (
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/testx"
 	"github.com/lf-edge/ekuiper/internal/topo/context"
+	"github.com/lf-edge/ekuiper/internal/topo/transform"
+	"github.com/lf-edge/ekuiper/pkg/cast"
 	"reflect"
 	"testing"
 )
@@ -36,8 +38,17 @@ var (
 
 func compareEvent(expected, actual *dtos.Event) bool {
 	if (expected.Id == actual.Id || (expected.Id == "" && actual.Id != "")) && expected.ProfileName == actual.ProfileName && expected.DeviceName == actual.DeviceName && (expected.Origin == actual.Origin || (expected.Origin == 0 && actual.Origin > 0)) && reflect.DeepEqual(expected.Tags, actual.Tags) && expected.SourceName == actual.SourceName && len(expected.Readings) == len(actual.Readings) {
-		for i, r := range expected.Readings {
-			if !compareReading(r, actual.Readings[i]) {
+		for _, r := range expected.Readings {
+			compared := false
+			for _, a := range actual.Readings {
+				if r.ResourceName == a.ResourceName {
+					compared = true
+					if !compareReading(r, a) {
+						return false
+					}
+				}
+			}
+			if !compared {
 				return false
 			}
 		}
@@ -262,8 +273,7 @@ func TestProduceEvents(t1 *testing.T) {
 
 		{ // 2
 			input: `[
-						{"meta": 50},
-						{"h1":100}
+						{"meta": 50,"h1":100}
 					]`,
 			conf: map[string]interface{}{
 				"sourceName": "demo",
@@ -488,3 +498,113 @@ func TestProduceEvents(t1 *testing.T) {
 		}
 	}
 }
+
+func TestEdgeXTemplate_Apply(t1 *testing.T) {
+	var tests = []struct {
+		input    string
+		conf     map[string]interface{}
+		expected *dtos.Event
+		error    string
+	}{
+		{ // 0
+			input: `[{"meta":{
+							"correlationid":"","deviceName":"demo","id":"","origin":3,
+							"humidity":{"deviceName":"test device name1","id":"12","origin":14,"valueType":"Int64"},
+							"temperature":{"deviceName":"test device name2","id":"22","origin":24}
+							},
+						"humidity":100,
+						"temperature":50}
+					]`,
+			conf: map[string]interface{}{
+				"metadata": "meta",
+
+				"dataTemplate": `{"wrapper":"w1","ab":"{{.humidity}}"}`,
+			},
+			expected: &dtos.Event{
+				Id:          "",
+				DeviceName:  "ekuiper",
+				ProfileName: "ekuiperProfile",
+				SourceName:  "ruleTest",
+				Origin:      0,
+				Readings: []dtos.BaseReading{
+					{
+						ResourceName:  "wrapper",
+						DeviceName:    "ekuiper",
+						ProfileName:   "ekuiperProfile",
+						Id:            "",
+						Origin:        0,
+						ValueType:     v2.ValueTypeString,
+						SimpleReading: dtos.SimpleReading{Value: "w1"},
+					},
+					{
+						ResourceName:  "ab",
+						DeviceName:    "ekuiper",
+						ProfileName:   "ekuiperProfile",
+						Id:            "",
+						Origin:        0,
+						ValueType:     v2.ValueTypeString,
+						SimpleReading: dtos.SimpleReading{Value: "100"},
+					},
+				},
+			},
+			error: "",
+		}, {
+			input: `[{"json":"{\"a\":24,\"b\":\"c\"}"}]`,
+			conf: map[string]interface{}{
+				"dataTemplate": `{{.json}}`,
+			},
+			expected: &dtos.Event{
+				Id:          "",
+				DeviceName:  "ekuiper",
+				ProfileName: "ekuiperProfile",
+				SourceName:  "ruleTest",
+				Origin:      0,
+				Readings: []dtos.BaseReading{
+					{
+						ResourceName:  "a",
+						DeviceName:    "ekuiper",
+						ProfileName:   "ekuiperProfile",
+						Id:            "",
+						Origin:        0,
+						ValueType:     v2.ValueTypeFloat64,
+						SimpleReading: dtos.SimpleReading{Value: "2.400000e+01"},
+					},
+					{
+						ResourceName:  "b",
+						DeviceName:    "ekuiper",
+						ProfileName:   "ekuiperProfile",
+						Id:            "",
+						Origin:        0,
+						ValueType:     v2.ValueTypeString,
+						SimpleReading: dtos.SimpleReading{Value: "c"},
+					},
+				},
+			},
+			error: "",
+		},
+	}
+
+	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
+	for i, t := range tests {
+		ems := EdgexMsgBusSink{}
+		err := ems.Configure(t.conf)
+		if err != nil {
+			t1.Errorf("%d: configure error %v", i, err)
+			continue
+		}
+		if ems.c.SourceName == "" {
+			ems.c.SourceName = "ruleTest"
+		}
+		var payload []map[string]interface{}
+		json.Unmarshal([]byte(t.input), &payload)
+		dt := t.conf["dataTemplate"]
+		tf, _ := transform.GenTransform(cast.ToStringAlways(dt))
+		vCtx := context.WithValue(ctx, context.TransKey, tf)
+		result, err := ems.produceEvents(vCtx, payload[0])
+		if !reflect.DeepEqual(t.error, testx.Errstring(err)) {
+			t1.Errorf("%d. %q: error mismatch:\n  exp=%s\n  got=%s\n\n", i, t.input, t.error, err)
+		} else if t.error == "" && !compareEvent(t.expected, result) {
+			t1.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, t.input, t.expected, result)
+		}
+	}
+}