Merge pull request #603 from emqx/dev/1.0.1

Release 1.0.1

.ci/Dockerfile-plugins

@@ -19,11 +19,14 @@ RUN set -e -u -x \
            ;; \
          tdengine ) \
            if [ "$(uname -m)" = "x86_64" ]; then \
-             wget "https://www.taosdata.com/download/download-all.php?pkgType=tdengine_linux&pkgName=TDengine-client-2.0.3.1-Linux-x64.tar.gz" -O /tmp/TDengine-client-2.0.3.1-Linux-x64.tar.gz \
-             && tar -zxvf /tmp/TDengine-client-2.0.3.1-Linux-x64.tar.gz \
-             && cd TDengine-client && ./install_client.sh && cd - \
-             && go build --buildmode=plugin -tags plugins -o plugins/$PLUGIN_TYPE/$PLUGIN_NAME/$PLUGIN_NAME@$VERSION.so plugins/$PLUGIN_TYPE/$PLUGIN_NAME/$PLUGIN_NAME.go; \
-           fi \
+             wget "https://www.taosdata.com/download/download-all.php?pkgType=tdengine_linux&pkgName=TDengine-client-2.0.6.0-Linux-x64.tar.gz" -O /tmp/TDengine-client-2.0.6.0.tar.gz; \
+           fi; \
+           if [ "$(uname -m)" = "aarch64" ]; then \
+             wget "https://www.taosdata.com/download/download-all.php?pkgType=tdengine_linux&pkgName=TDengine-client-2.0.6.0-Linux-aarch64.tar.gz" -O /tmp/TDengine-client-2.0.6.0.tar.gz; \
+           fi; \
+           tar -zxvf /tmp/TDengine-client-2.0.6.0.tar.gz \
+           && cd TDengine-client-2.0.6.0 && ./install_client.sh && cd - \
+           && go build --buildmode=plugin -tags plugins -o plugins/$PLUGIN_TYPE/$PLUGIN_NAME/$PLUGIN_NAME@$VERSION.so plugins/$PLUGIN_TYPE/$PLUGIN_NAME/$PLUGIN_NAME.go \
            ;; \
          * ) \
            go build --buildmode=plugin -o plugins/$PLUGIN_TYPE/$PLUGIN_NAME/$PLUGIN_NAME@$VERSION.so plugins/$PLUGIN_TYPE/$PLUGIN_NAME/$PLUGIN_NAME.go \

.github/workflows/build_packages.yaml

@@ -262,13 +262,12 @@ jobs:
             cd _packages && for var in $( ls |grep -v sha256); do
               echo "$(cat $var.sha256) $var" | sha256sum -c || exit 1
             done
-        - name: update github release
+        - uses: zhanghongtong/upload-release-asset@v1
           if: github.event_name == 'release'
-          run: |
-            version=$(echo ${{ github.ref }} | sed -r  "s .*/.*/(.*) \1 g")
-            for var in $(ls _packages) ; do
-              .github/workflows/script/upload_github_release_asset.sh owner=emqx repo=kuiper tag=$version filename=_packages/$var github_api_token=$(echo ${{ secrets.AccessToken }})
-            done
+          with:
+            repo: kuiper
+            path: "_packages/kuiper-*"
+            token: ${{ secrets.AccessToken }}
         - name: create invalidation for cloudfront
           if: github.event_name == 'release'
           run: |
@@ -287,7 +286,13 @@ jobs:
           if: github.event_name == 'release'
           run: |
             version=$(echo ${{ github.ref }} | sed -r  "s .*/.*/(.*) \1 g")
-            curl -w %{http_code} --insecure -H ${{ secrets.EmqxHeader }} https://admin.emqx.io/admin_api/v1/kuiper_github_release_callback?tag=$version
+            curl -w %{http_code} \
+               --insecure \
+               -H "Content-Type: application/json" \
+               -H "token: ${{ secrets.EMQX_IO_TOKEN }}" \
+               -X POST \
+               -d "{\"repo\":\"emqx/kuiper\", \"tag\": \"${version}\" }" \
+               ${{ secrets.EMQX_IO_RELEASE_API }}
         - name: update helm packages
           if: github.event_name == 'release'
           run: |

.github/workflows/run_fvt_tests.yaml

@@ -23,9 +23,9 @@ jobs:
         - name: install jmeter
           timeout-minutes: 10
           env:
-            JMETER_VERSION: 5.2.1
+              JMETER_VERSION: 5.3
           run: |
-            wget --no-check-certificate -O /tmp/apache-jmeter.tgz http://us.mirrors.quenda.co/apache//jmeter/binaries/apache-jmeter-$JMETER_VERSION.tgz --no-check-certificate
+            wget --no-check-certificate -O /tmp/apache-jmeter.tgz https://downloads.apache.org//jmeter/binaries/apache-jmeter-$JMETER_VERSION.tgz
             cd /tmp && tar -xvf apache-jmeter.tgz
             echo "jmeter.save.saveservice.output_format=xml" >> /tmp/apache-jmeter-$JMETER_VERSION/user.properties
             echo "jmeter.save.saveservice.response_data.on_error=true" >> /tmp/apache-jmeter-$JMETER_VERSION/user.properties
@@ -84,9 +84,9 @@ jobs:
       - name: install jmeter
         timeout-minutes: 10
         env:
-          JMETER_VERSION: 5.2.1
+            JMETER_VERSION: 5.3
         run: |
-          wget --no-check-certificate -O /tmp/apache-jmeter.tgz http://us.mirrors.quenda.co/apache//jmeter/binaries/apache-jmeter-$JMETER_VERSION.tgz
+          wget --no-check-certificate -O /tmp/apache-jmeter.tgz https://downloads.apache.org//jmeter/binaries/apache-jmeter-$JMETER_VERSION.tgz
           cd /tmp && tar -xvf apache-jmeter.tgz
           echo "jmeter.save.saveservice.output_format=xml" >> /tmp/apache-jmeter-$JMETER_VERSION/user.properties
           echo "jmeter.save.saveservice.response_data.on_error=true" >> /tmp/apache-jmeter-$JMETER_VERSION/user.properties

Makefile

@@ -182,7 +182,7 @@ PLUGINS := sinks/file \
 plugins: cross_prepare sinks/tdengine $(PLUGINS)
 sinks/tdengine:
 	@docker buildx build --no-cache \
-    --platform=linux/amd64 \
+    --platform=linux/amd64,linux/arm64 \
     -t cross_build \
     --build-arg VERSION=$(VERSION) \
     --build-arg PLUGIN_TYPE=sinks \
@@ -191,9 +191,12 @@ sinks/tdengine:
     -f .ci/Dockerfile-plugins .
 
 	@mkdir -p _plugins/debian/sinks
-	@tar -xvf /tmp/cross_build_plugins_sinks_tdengine.tar --wildcards "go/kuiper/plugins/sinks/tdengine/tdengine_amd64.zip" \
-	&& mv go/kuiper/plugins/sinks/tdengine/tdengine_amd64.zip _plugins/debian/sinks
+	@for arch in amd64 arm64; do \
+		tar -xvf /tmp/cross_build_plugins_sinks_tdengine.tar --wildcards "linux_$${arch}/go/kuiper/plugins/sinks/tdengine/tdengine_$$(echo $${arch%%_*}).zip" \
+		&& mv $$(ls linux_$${arch}/go/kuiper/plugins/sinks/tdengine/tdengine_$$(echo $${arch%%_*}).zip) _plugins/debian/sinks; \
+	done
 	@rm -f /tmp/cross_build_plugins_sinks_tdengine.tar
+
 $(PLUGINS): PLUGIN_TYPE = $(word 1, $(subst /, , $@))
 $(PLUGINS): PLUGIN_NAME = $(word 2, $(subst /, , $@))
 $(PLUGINS):

common/util.go

@@ -15,19 +15,22 @@ import (
 	"path"
 	"path/filepath"
 	//"runtime"
+	logrus_syslog "github.com/sirupsen/logrus/hooks/syslog"
+	"log/syslog"
 	"sort"
 	"strings"
 	"sync"
 )
 
 const (
-	logFileName   = "stream.log"
-	etc_dir       = "/etc/"
-	data_dir      = "/data/"
-	log_dir       = "/log/"
-	StreamConf    = "kuiper.yaml"
-	KuiperBaseKey = "KuiperBaseKey"
-	MetaKey       = "__meta"
+	logFileName     = "stream.log"
+	etc_dir         = "/etc/"
+	data_dir        = "/data/"
+	log_dir         = "/log/"
+	StreamConf      = "kuiper.yaml"
+	KuiperBaseKey   = "KuiperBaseKey"
+	KuiperSyslogKey = "KuiperSyslogKey"
+	MetaKey         = "__meta"
 )
 
 var (
@@ -79,21 +82,19 @@ type KuiperConf struct {
 }
 
 func init() {
+	Log = logrus.New()
+	if "true" == os.Getenv(KuiperSyslogKey) {
+		if hook, err := logrus_syslog.NewSyslogHook("", "", syslog.LOG_INFO, ""); err != nil {
+			Log.Error("Unable to connect to local syslog daemon")
+		} else {
+			Log.AddHook(hook)
+		}
+	}
+
 	filenameHook := filename.NewHook()
 	filenameHook.Field = "file"
-	Log = logrus.New()
 	Log.AddHook(filenameHook)
-	/*
-		Log.SetReportCaller(true)
-			Log.SetFormatter(&logrus.TextFormatter{
-				CallerPrettyfier: func(f *runtime.Frame) (string, string) {
-					filename := path.Base(f.File)
-					return "", fmt.Sprintf("%s:%d", filename, f.Line)
-				},
-				DisableColors: true,
-				FullTimestamp: true,
-			})
-	*/
+
 	Log.SetFormatter(&logrus.TextFormatter{
 		TimestampFormat: "2006-01-02 15:04:05",
 		DisableColors:   true,

docs/en_US/edgex/edgex_rule_engine_tutorial.md

@@ -71,6 +71,10 @@ f69e9c4d6cc8        nexus3.edgexfoundry.org:10004/docker-core-data-go:master
 ed7ad5ae08b2        nexus3.edgexfoundry.org:10004/docker-edgex-volume:master               "/bin/sh -c '/usr/bi…"   37 minutes ago      Up 37 minutes                                                                                                          edgex-files
 ```
 
+#### Run with native
+
+For performance reason, reader probably wants to run Kuiper with native approach. But you may find that [EdgeX cannot be used](https://github.com/emqx/kuiper/issues/596) with the downloaded Kuiper binary packages. It's because that EdgeX message bus relies on `zeromq` library. If  `zeromq` library cannot be found in the library search path, it cannot be started. So it will have those Kuiper users who do not want to use EdgeX install the `zeromq` library as well. For this reason, the default downloaded Kuiper package **<u>does NOT have embedded support</u>** for `EdgeX`. If reader wants to support `EdgeX` in native packages, you can either make a native package by running command `make pkg_with_edgex`, or just copy the binary package from docker container.
+
 ### Create a stream
 
 There are two approaches to manage stream, you can use your preferred approach.

docs/en_US/operation/configuration_file.md

@@ -12,7 +12,8 @@ basic:
   # true|false, if it's set to true, then the log will be print to log file
   fileLog: true
 ```
-
+## system log
+When the user sets the value of the environment variable named KuiperSyslogKey to true, the log will be printed to the syslog.
 ## Cli Port
 ```yaml
 basic:

docs/zh_CN/edgex/edgex_rule_engine_tutorial.md

@@ -67,6 +67,10 @@ f69e9c4d6cc8        nexus3.edgexfoundry.org:10004/docker-core-data-go:master
 ed7ad5ae08b2        nexus3.edgexfoundry.org:10004/docker-edgex-volume:master               "/bin/sh -c '/usr/bi…"   37 minutes ago      Up 37 minutes                                                                                                          edgex-files
 ```
 
+#### 原生 (native) 方式运行
+
+出于运行效率考虑，读者可能需要直接以原生方式运行 Kuiper，但是可能会发现直接使用下载的 Kuiper 软件包启动后[无法直接使用 Edgex](https://github.com/emqx/kuiper/issues/596)，这是因为 EdgeX 缺省消息总线依赖于 `zeromq` 库，如果 Kuiper 启动的时候在库文件寻找路径下无法找到 `zeromq` 库，它将无法启动。这导致对于不需要使用 EdgeX 的 Kuiper 用户也不得不去安装 `zeromq` 库 ，因此缺省提供的下载安装包中**<u>内置不支持 Edgex</u>** 。如果读者需要以原生方式运行 Kuiper 并且支持 `EdgeX`，可以通过命令 `make pkg_with_edgex` 自己来编译原生安装包，或者从容器中直接拷贝出安装包。
+
 ### 创建流
 
 该步骤是创建一个可以从 EdgeX 消息总线进行数据消费的流。有两种方法来支持管理流，你可以选择喜欢的方式。

docs/zh_CN/operation/configuration_file.md

@@ -12,6 +12,8 @@ basic:
   # true|false, if it's set to true, then the log will be print to log file
   fileLog: true
 ```
+## 系统日志
+用户将名为 KuiperSyslogKey 的环境变量的值设置为 true 时，日志将打印到系统日志中。
 ## Cli 端口
 ```yaml
 basic:

etc/sinks/rest.json

@@ -95,10 +95,10 @@
     },
     {
       "name": "headers",
-      "default": "",
+      "default": [],
       "optional": true,
-      "control": "text",
-      "type": "string",
+      "control": "list",
+      "type": "list_object",
       "hint": {
         "en_US": "The additional headers to be set for the HTTP request.",
         "zh_CN": "要为 HTTP 请求设置的其他标头"

plugins/sinks/tdengine/install.sh

@@ -8,6 +8,9 @@ then
 fi
 
 url="https://www.taosdata.com/download/download-all.php?pkgType=tdengine_linux&pkgName=TDengine-client-$1-Linux-x64.tar.gz"
+if [ "$(uname -m)" = "aarch64" ]; then \
+	url="https://www.taosdata.com/download/download-all.php?pkgType=tdengine_linux&pkgName=TDengine-client-$1-Linux-aarch64.tar.gz"
+fi
 zip="TDengine-client.tar.gz"
 wget -T 280 -O "$zip" "$url"
 
@@ -18,7 +21,8 @@ then
 fi
 
 dir="TDengine-client"
-tar -zxvf "$zip"
+mkdir "$dir"
+tar -xzvf "$zip" -C ./"$dir" --strip-components 1
 rm "$zip"
 
 if ! [ -e $dir ]

xsql/ast.go

@@ -314,10 +314,10 @@ type StreamField struct {
 
 func (u *StreamField) MarshalJSON() ([]byte, error) {
 	return json.Marshal(&struct {
-		FieldType string
+		FieldType interface{}
 		Name      string
 	}{
-		FieldType: PrintFieldType(u.FieldType),
+		FieldType: PrintFieldTypeForJson(u.FieldType),
 		Name:      u.Name,
 	})
 }

xsql/ast_test.go

@@ -1,6 +1,7 @@
 package xsql
 
 import (
+	"encoding/json"
 	"fmt"
 	"reflect"
 	"testing"
@@ -108,3 +109,42 @@ func Test_MessageValTest(t *testing.T) {
 		}
 	}
 }
+
+func Test_StreamFieldsMarshall(t *testing.T) {
+	var tests = []struct {
+		sf StreamFields
+		r  string
+	}{{
+		sf: []StreamField{
+			{Name: "USERID", FieldType: &BasicType{Type: BIGINT}},
+			{Name: "FIRST_NAME", FieldType: &BasicType{Type: STRINGS}},
+			{Name: "LAST_NAME", FieldType: &BasicType{Type: STRINGS}},
+			{Name: "NICKNAMES", FieldType: &ArrayType{Type: STRINGS}},
+			{Name: "Gender", FieldType: &BasicType{Type: BOOLEAN}},
+			{Name: "ADDRESS", FieldType: &RecType{
+				StreamFields: []StreamField{
+					{Name: "STREET_NAME", FieldType: &BasicType{Type: STRINGS}},
+					{Name: "NUMBER", FieldType: &BasicType{Type: BIGINT}},
+				},
+			}},
+		},
+		r: `[{"FieldType":"bigint","Name":"USERID"},{"FieldType":"string","Name":"FIRST_NAME"},{"FieldType":"string","Name":"LAST_NAME"},{"FieldType":{"Type":"array","ElementType":"string"},"Name":"NICKNAMES"},{"FieldType":"boolean","Name":"Gender"},{"FieldType":{"Type":"struct","Fields":[{"FieldType":"string","Name":"STREET_NAME"},{"FieldType":"bigint","Name":"NUMBER"}]},"Name":"ADDRESS"}]`,
+	}, {
+		sf: []StreamField{
+			{Name: "USERID", FieldType: &BasicType{Type: BIGINT}},
+		},
+		r: `[{"FieldType":"bigint","Name":"USERID"}]`,
+	}}
+	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
+	for i, tt := range tests {
+		r, err := json.Marshal(tt.sf)
+		if err != nil {
+			t.Errorf("%d. \nmarshall error: %v", i, err)
+			t.FailNow()
+		}
+		result := string(r)
+		if !reflect.DeepEqual(tt.r, result) {
+			t.Errorf("%d. \nstmt mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.r, result)
+		}
+	}
+}

xsql/processors/common_test.go

@@ -71,7 +71,7 @@ func compareMetrics(tp *xstream.TopologyNew, m map[string]interface{}) (err erro
 		)
 		for index, key = range keys {
 			if k == key {
-				if strings.HasSuffix(k, "process_latency_ms") {
+				if strings.HasSuffix(k, "process_latency_us") {
 					if values[index].(int64) >= v.(int64) {
 						matched = true
 						continue