Repository: lf-edge/ekuiper
Pull request #1296 (2022-06-10 04:05:49)

fix(doc): escape for {{}}


Patch:
diff --git a/docs/en_US/rules/sinks/builtin/edgex.md b/docs/en_US/rules/sinks/builtin/edgex.md
index b8284e04..5714d88f 100644
--- a/docs/en_US/rules/sinks/builtin/edgex.md
+++ b/docs/en_US/rules/sinks/builtin/edgex.md
@@ -39,7 +39,9 @@ Below optional configurations are supported, please check MQTT specification for
   - KeyPEMBlock
   - SkipCertVerify
 
+::: v-pre
 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
 
diff --git a/docs/en_US/rules/sinks/builtin/memory.md b/docs/en_US/rules/sinks/builtin/memory.md
index 5bea5dca..0c8848c9 100644
--- a/docs/en_US/rules/sinks/builtin/memory.md
+++ b/docs/en_US/rules/sinks/builtin/memory.md
@@ -28,4 +28,6 @@ Below is another sample for dynamic topic action:
 
 ## Data Templates
 
-The data transfer between the memory action and the memory source is in internal format and is not coded or decoded for efficiency. Therefore, the format-related configuration items of the memory action are ignored, except for the data template. The memory 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.
\ No newline at end of file
+::: v-pre
+The data transfer between the memory action and the memory source is in internal format and is not coded or decoded for efficiency. Therefore, the format-related configuration items of the memory action are ignored, except for the data template. The memory 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.
+:::
\ 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 7246fa0b..8c120200 100644
--- a/docs/zh_CN/rules/sinks/builtin/edgex.md
+++ b/docs/zh_CN/rules/sinks/builtin/edgex.md
@@ -39,7 +39,9 @@
   - KeyPEMBlock
   - SkipCertVerify
 
+::: v-pre
 EdgeX 动作可支持数据模板对结果格式进行变化，但是数据模板的结果必须为 JSON 字符串的 object 形式，例如 `"{\"key\":\"{{.key}}\"}"`。数组形式的 JSON 字符串或者非 JSON 字符串都不支持。
+:::
 
 ## 发送到各种目标
 
diff --git a/docs/zh_CN/rules/sinks/builtin/memory.md b/docs/zh_CN/rules/sinks/builtin/memory.md
index 1d2da957..47dd7ce4 100644
--- a/docs/zh_CN/rules/sinks/builtin/memory.md
+++ b/docs/zh_CN/rules/sinks/builtin/memory.md
@@ -28,4 +28,6 @@
 
 ## 数据模板
 
-内存动作和内存源之间的数据传输采用内部格式，不经过编解码以提高效率。因此，内存动作的格式相关配置项，除了数据模板之外都会被忽略。内存动作可支持数据模板对结果格式进行变化，但是数据模板的结果必须为 JSON 字符串的 object 形式，例如 `"{\"key\":\"{{.key}}\"}"`。数组形式的 JSON 字符串或者非 JSON 字符串都不支持。
\ No newline at end of file
+::: v-pre
+内存动作和内存源之间的数据传输采用内部格式，不经过编解码以提高效率。因此，内存动作的格式相关配置项，除了数据模板之外都会被忽略。内存动作可支持数据模板对结果格式进行变化，但是数据模板的结果必须为 JSON 字符串的 object 形式，例如 `"{\"key\":\"{{.key}}\"}"`。数组形式的 JSON 字符串或者非 JSON 字符串都不支持。
+:::
\ No newline at end of file
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


Test patch:
diff --git a/internal/topo/sink/edgex_sink_test.go b/internal/topo/sink/edgex_sink_test.go
index 255c258b..4f5cc6e0 100644
--- a/internal/topo/sink/edgex_sink_test.go
+++ b/internal/topo/sink/edgex_sink_test.go
@@ -38,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
 			}
 		}