Repository: lf-edge/ekuiper
Pull request #1371 (2022-08-16 09:41:28)

fix(ci): build plugins with alpine build image
Signed-off-by: Jianxiang Ran <rxan_embedded@163.com>


Patch:
diff --git a/.github/workflows/build_packages.yaml b/.github/workflows/build_packages.yaml
index 277f2255..bbd58bb5 100644
--- a/.github/workflows/build_packages.yaml
+++ b/.github/workflows/build_packages.yaml
@@ -195,11 +195,18 @@ jobs:
         arch:
           - linux/amd64
           - linux/arm64
+        os:
+          - [debian, "slim"]
+          - [alpine, "alpine"]
         golang:
           - 1.18.5
         exclude:
           - arch: linux/arm64
             plugin: functions/labelImage
+          - os: [alpine,"alpine"]
+            plugin: functions/labelImage
+          - os: [alpine,"alpine"]
+            plugin: sinks/tdengine
 
     steps:
     - uses: actions/checkout@v2
@@ -210,13 +217,13 @@ jobs:
       with:
         image: tonistiigi/binfmt:latest
         platforms: all
-    - name: build debian plugins
+    - name: build plugins
       run: |
         docker run -i --rm \
             -v $(pwd):/ekuiper \
             --workdir /ekuiper \
             --platform ${{ matrix.arch }} \
-            ghcr.io/lf-edge/ekuiper/base:${{ matrix.golang }}-debian \
+            ghcr.io/lf-edge/ekuiper/base:${{ matrix.golang }}-${{ matrix.os[0] }} \
             bash -euc "make ${{ matrix.plugin }}"
     - name: Build ekuiper image
       uses: docker/build-push-action@v2
@@ -226,8 +233,8 @@ jobs:
         platforms: linux/amd64
         push: false
         load: true
-        tags: docker.io/lfedge/ekuiper
-        file: deploy/docker/Dockerfile-slim
+        tags: docker.io/lfedge/ekuiper-${{ matrix.os[0] }}
+        file: deploy/docker/Dockerfile-${{ matrix.os[1] }}
     - name: test docker and plugins
       if: matrix.arch == 'linux/amd64'
       env:
@@ -237,7 +244,7 @@ jobs:
         plugin_type=$(echo ${PLUGIN%%/*})
         plugin_name=$(echo ${PLUGIN##*/})
 
-        container_id=$(docker run -d -v $(pwd)/_plugins:/var/plugins docker.io/lfedge/ekuiper)
+        container_id=$(docker run -d -v $(pwd)/_plugins:/var/plugins docker.io/lfedge/ekuiper-${{ matrix.os[0] }})
         ip_address=$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' $container_id)
         os=$(docker exec -i ${container_id} sh -c "sed -n '/^ID=/p' /etc/os-release | sed -r 's/ID=(.*)/\1/g'" )
         sleep 5
diff --git a/build-plugins.sh b/build-plugins.sh
index 02dda999..ae8c5809 100755
--- a/build-plugins.sh
+++ b/build-plugins.sh
@@ -6,9 +6,10 @@ cd -P -- "$(dirname -- "${BASH_SOURCE[0]}")"
 PLUGIN_TYPE=$1
 PLUGIN_NAME=$2
 VERSION=$(git describe --tags --always)
+OS=$(sed -n '/^ID=/p' /etc/os-release | sed -r 's/ID=(.*)/\1/g')
 
 pre(){
-    mkdir -p _plugins/debian/$PLUGIN_TYPE
+    mkdir -p _plugins/$OS/$PLUGIN_TYPE
     if [  $(cat etc/$PLUGIN_TYPE/$PLUGIN_NAME.json | jq -r ".libs") != 'null' ]; then
         for lib in $(cat etc/$PLUGIN_TYPE/$PLUGIN_NAME.json | jq -r ".libs[]"); do
             go get $lib;
@@ -23,7 +24,7 @@ post(){
     cd extensions/$PLUGIN_TYPE/$PLUGIN_NAME
     zip -r ${PLUGIN_NAME}_$(go env GOARCH).zip .
     cd -
-    mv $(find extensions/$PLUGIN_TYPE/$PLUGIN_NAME -name "*.zip")  _plugins/debian/$PLUGIN_TYPE
+    mv $(find extensions/$PLUGIN_TYPE/$PLUGIN_NAME -name "*.zip")  _plugins/$OS/$PLUGIN_TYPE
 }
 
 build(){
diff --git a/deploy/docker/Dockerfile b/deploy/docker/Dockerfile
index 07752a8a..d1b62671 100644
--- a/deploy/docker/Dockerfile
+++ b/deploy/docker/Dockerfile
@@ -19,9 +19,7 @@ COPY . /go/kuiper
 
 WORKDIR /go/kuiper
 
-RUN apt-get update \
-    && apt-get install -y pkg-config libczmq-dev wget \
-    && apt-get clean \
+RUN apt-get clean \
     && rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/*
 
 RUN make build_with_edgex \
diff --git a/deploy/docker/Dockerfile-alpine b/deploy/docker/Dockerfile-alpine
index 15a10318..2f7c80d5 100644
--- a/deploy/docker/Dockerfile-alpine
+++ b/deploy/docker/Dockerfile-alpine
@@ -13,13 +13,12 @@
 # limitations under the License.
 
 ARG GO_VERSION=1.18.5
-FROM golang:$GO_VERSION-alpine AS builder
+FROM ghcr.io/lf-edge/ekuiper/base:$GO_VERSION-alpine AS builder
 
 COPY . /go/kuiper
 
 WORKDIR /go/kuiper
 
-RUN apk add gcc make git libc-dev binutils-gold pkgconfig zeromq-dev 
 RUN make build_with_edgex
 
 FROM alpine:3.15
diff --git a/deploy/docker/Dockerfile-slim b/deploy/docker/Dockerfile-slim
index 83a74a0b..bae98338 100644
--- a/deploy/docker/Dockerfile-slim
+++ b/deploy/docker/Dockerfile-slim
@@ -19,7 +19,6 @@ COPY . /go/kuiper
 
 WORKDIR /go/kuiper
 
-RUN apt update && apt install -y pkg-config libczmq-dev
 RUN make build_with_edgex
 
 FROM debian:buster-20211011 
diff --git a/deploy/docker/Dockerfile-slim-python b/deploy/docker/Dockerfile-slim-python
index 83bf7e85..ccd4d7c1 100644
--- a/deploy/docker/Dockerfile-slim-python
+++ b/deploy/docker/Dockerfile-slim-python
@@ -19,7 +19,6 @@ COPY . /go/kuiper
 
 WORKDIR /go/kuiper
 
-RUN apt update && apt install -y pkg-config libczmq-dev
 RUN make build_with_edgex
 
 FROM python:3.8.12-slim-bullseye
diff --git a/docs/en_US/rules/sinks/builtin/neuron.md b/docs/en_US/rules/sinks/builtin/neuron.md
index 312fb750..ac7fc827 100644
--- a/docs/en_US/rules/sinks/builtin/neuron.md
+++ b/docs/en_US/rules/sinks/builtin/neuron.md
@@ -1,6 +1,6 @@
 # Neuron action
 
-The action is used to publish result to the local neuron instance. Notice that, the sink is bound to the local neuron only which must be able to communicate through nanomsg ipc protocol without network. In the eKuiper side, all neuron source and sink instances share the same connection.
+The action is used to publish result to the local neuron instance. Notice that, the sink is bound to the local neuron only which must be able to communicate through nanomsg ipc protocol without network. In the eKuiper side, all neuron source and sink instances share the same connection. Notice that, the dial to neuron is async which will run in the background and always redial when the previous attempt fails, which means that the rule using neuron sink will not see an error even when neuron is down. Additionally, there is a send queue of 128 messages in the nanomsg client, thus the neuron sink can have 128 messages send out even when neuron is down. The rule will only start to get timeout errors after 128 queued messages. Once the connection is restored, the queued messages will be sent out automatically.
 
 | Property name | Optional | Description                                                                                                                             |
 |---------------|----------|-----------------------------------------------------------------------------------------------------------------------------------------|
diff --git a/docs/en_US/rules/sources/builtin/neuron.md b/docs/en_US/rules/sources/builtin/neuron.md
index 04bfa2fc..74b55179 100644
--- a/docs/en_US/rules/sources/builtin/neuron.md
+++ b/docs/en_US/rules/sources/builtin/neuron.md
@@ -1,6 +1,8 @@
 # Neuron Source
 
-Neuron source is provided to consume events from the local neuron instance. Notice that, the source is bound to the local neuron only which must be able to communicate through nanomsg ipc protocol without network. In the eKuiper side, all neuron source and sink instances share the same connection, thus the events consumed are also the same. The events are formatted in json by default. The event content and format are fixed as below: 
+Neuron source is provided to consume events from the local neuron instance. Notice that, the source is bound to the local neuron only which must be able to communicate through nanomsg ipc protocol without network. In the eKuiper side, all neuron source and sink instances share the same connection, thus the events consumed are also the same. The events are formatted in json by default. Notice that, the dial to neuron is async which will run in the background and always redial when the previous attempt fails, which means that the rule using neuron sink will not see an error even when neuron is down. While debugging, we need to check the rule status for the message in counts.
+
+The event content and format are fixed as below: 
 
 ```json
 {
diff --git a/docs/en_US/sqls/streams.md b/docs/en_US/sqls/streams.md
index d650bd60..254455ec 100644
--- a/docs/en_US/sqls/streams.md
+++ b/docs/en_US/sqls/streams.md
@@ -31,12 +31,11 @@ CREATE STREAM
 | Property name | Optional | Description                                                                                                                                                |
 |---------------|----------|------------------------------------------------------------------------------------------------------------------------------------------------------------|
 | DATASOURCE    | false    | The value is determined by source type. The topic names list if it's a MQTT data source. Please refer to related document for other sources.               |
-| FORMAT        | true     | The data format, currently the value can be "JSON", "PROTOBUF" and "BINARY". The default is "JSON". Check [Binary Stream](#Binary Stream) for more detail. |
+| FORMAT        | true     | The data format, currently the value can be "JSON", "PROTOBUF" and "BINARY". The default is "JSON". Check [Binary Stream](#binary-stream) for more detail. |
 | SCHEMAID      | true     | The schema to be used when decoding the events. Currently, only use when format is PROTOBUF.                                                               |
-
 | KEY              | true     | Reserved key, currently the field is not used. It will be used for GROUP BY statements.                                                                                                                                            |
 | TYPE             | true     | The source type, if not specified, the value is "mqtt".                                                                                                                                                                            |
-| StrictValidation | true     | To control validation behavior of message field against stream schema. See [Strict Validation](#Strict Validation) for more info.                                                                                                  |
+| StrictValidation | true     | To control validation behavior of message field against stream schema. See [Strict Validation](#strict-validation) for more info.                                                                                                  |
 | CONF_KEY         | true     | If additional configuration items are requied to be configured, then specify the config key here. See [MQTT stream](../rules/sources/builtin/mqtt.md) for more info.                                                               |
 | SHARED           | true     | Whether the source instance will be shared across all rules using this stream                                                                                                                                                      |
 | TIMESTAMP        | true     | The field to represent the event's timestamp. If specified, the rule will run with event time. Otherwise, it will run with processing time. Please refer to [timestamp management](./windows.md#timestamp-management) for details. |
diff --git a/docs/zh_CN/rules/sinks/builtin/neuron.md b/docs/zh_CN/rules/sinks/builtin/neuron.md
index b745171f..e9f8c305 100644
--- a/docs/zh_CN/rules/sinks/builtin/neuron.md
+++ b/docs/zh_CN/rules/sinks/builtin/neuron.md
@@ -1,6 +1,6 @@
 # Neuron 动作
 
-该动作用于将结果发送到本地的 neuron 实例中以反控 neuron。需要注意的是，该动作与本地的 neuron 绑定，因为通信是通过 nanomsg 的 ipc 协议进行，无法通过网络通信。在 eKuiper 端，所有的 neuron 源和动作共享同一个全局 neuron 连接。
+该动作用于将结果发送到本地的 neuron 实例中以反控 neuron。需要注意的是，该动作与本地的 neuron 绑定，因为通信是通过 nanomsg 的 ipc 协议进行，无法通过网络通信。在 eKuiper 端，所有的 neuron 源和动作共享同一个全局 neuron 连接。需要注意的是，拨号到 Neuron 是异步的，它将在后台运行，不断重拨直到连接成功。这意味着使用 Neuron sink 的规则即使在 Neuron 停机时也不会看到错误。此外，在 nanomsg 消息客户端中有一个 128 条消息的发送队列，因此即使在 Neuron 停机时，Neuron sink 也能有 128 条消息发送出去。规则只有在 128 条消息队列满之后才会开始出现超时错误。一旦连接恢复，队列中的消息就会自动发送出去。
 
 | 属性名称      | 是否可选 | 描述                                                                   |
 |-----------|------|----------------------------------------------------------------------|
diff --git a/docs/zh_CN/rules/sources/builtin/neuron.md b/docs/zh_CN/rules/sources/builtin/neuron.md
index 56105ff2..6ce10be3 100644
--- a/docs/zh_CN/rules/sources/builtin/neuron.md
+++ b/docs/zh_CN/rules/sources/builtin/neuron.md
@@ -1,6 +1,8 @@
 # Neuron 源
 
-Neuron 源订阅本地 neuron 实例的消息。需要注意的是，该源仅可用于本地的 neuron，因为与 neuron 的通信基于 nanomsg ipc 协议，无法通过网络进行。在 eKuiper 端，所有 neuron 源和动作共享同一个 neuron 连接。Neuron 发过来的消息为固定的 json 格式，如下所示： 
+Neuron 源订阅本地 neuron 实例的消息。需要注意的是，该源仅可用于本地的 neuron，因为与 neuron 的通信基于 nanomsg ipc 协议，无法通过网络进行。在 eKuiper 端，所有 neuron 源和动作共享同一个 neuron 连接。需要注意的是，拨号到 Neuron 是异步的，它将在后台运行，不断重拨直到连接成功。这意味着使用 Neuron sink 的规则即使在 Neuron 停机时也不会看到错误。用户调试时，可查看规则状态，注意消息流入数量是否正常。
+
+Neuron 发过来的消息为固定的 json 格式，如下所示： 
 
 ```json
 {
diff --git a/docs/zh_CN/sqls/streams.md b/docs/zh_CN/sqls/streams.md
index e31ab2f8..d3ef7a8c 100755
--- a/docs/zh_CN/sqls/streams.md
+++ b/docs/zh_CN/sqls/streams.md
@@ -37,7 +37,7 @@ CREATE STREAM
 | SCHEMAID         | 是   | 解码时使用的模式，目前仅在格式为 PROTOBUF 的情况下使用。                                                                                                                                       |
 | KEY              | 是   | 保留配置，当前未使用该字段。 它将用于 GROUP BY 语句。                                                                                                                                        |
 | TYPE             | 是   | 源类型，如未指定，值为 "mqtt"。                                                                                                                                                     |
-| StrictValidation | 是   | 针对流模式控制消息字段的验证行为。 有关更多信息，请参见 [Strict Validation](#Strict Validation)                                                                                                    |
+| StrictValidation | 是   | 针对流模式控制消息字段的验证行为。 有关更多信息，请参见 [Strict Validation](#strict-validation)                                                                                                    |
 | CONF_KEY         | 是   | 如果需要配置其他配置项，请在此处指定 config 键。 有关更多信息，请参见 [MQTT stream](../rules/sources/builtin/mqtt.md) 。                                                                               |
 | SHARED           | 是   | 是否在使用该流的规则中共享源的实例                                                                                                                                                       |
 | TIMESTAMP        | 是   | 代表该事件时间戳的字段名。如果有设置，则使用此流的规则将采用事件时间；否则将采用处理时间。详情请看[时间戳管理](./windows.md#时间戳管理)。                                                                                           |
@@ -134,4 +134,4 @@ demoBin (
 ) WITH (DATASOURCE="test/", FORMAT="BINARY");
 ```
 
-如果 "BINARY" 格式流定义为 schemaless，数据将会解析到默认的名为 `self` 的字段。
\ No newline at end of file
+如果 "BINARY" 格式流定义为 schemaless，数据将会解析到默认的名为 `self` 的字段。
diff --git a/internal/server/meta_plugin_init.go b/internal/server/meta_plugin_init.go
index 82e4ac12..d5fec378 100644
--- a/internal/server/meta_plugin_init.go
+++ b/internal/server/meta_plugin_init.go
@@ -24,7 +24,6 @@ import (
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/plugin"
 	"net/http"
-	"os"
 	"runtime"
 	"strings"
 )
@@ -50,37 +49,25 @@ func prebuildFuncsPlugins(w http.ResponseWriter, r *http.Request) {
 	prebuildPluginsHandler(w, r, plugin.FUNCTION)
 }
 
-func isOffcialDockerImage() bool {
-	if !strings.EqualFold(os.Getenv("MAINTAINER"), "emqx.io") {
-		return false
-	}
-	return true
-}
-
-func prebuildPluginsHandler(w http.ResponseWriter, r *http.Request, t plugin.PluginType) {
-	emsg := "It's strongly recommended to install plugins at official released Debian Docker images. If you choose to proceed to install plugin, please make sure the plugin is already validated in your own build."
-	if !isOffcialDockerImage() {
-		handleError(w, fmt.Errorf(emsg), "", logger)
-		return
-	} else if runtime.GOOS == "linux" {
+func prebuildPluginsHandler(w http.ResponseWriter, _ *http.Request, t plugin.PluginType) {
+	emsg := "It's strongly recommended to install plugins at linux. If you choose to proceed to install plugin, please make sure the plugin is already validated in your own build."
+	if runtime.GOOS == "linux" {
 		osrelease, err := Read()
 		if err != nil {
 			handleError(w, err, "", logger)
 			return
 		}
 		prettyName := strings.ToUpper(osrelease["PRETTY_NAME"])
-		os := "debian"
-		if strings.Contains(prettyName, "DEBIAN") {
-			hosts := conf.Config.Basic.PluginHosts
+		var os = "debian"
+		if strings.Contains(prettyName, "ALPINE") {
+			os = "alpine"
+		}
 
-			if err, plugins := fetchPluginList(t, hosts, os, runtime.GOARCH); err != nil {
-				handleError(w, err, "", logger)
-			} else {
-				jsonResponse(plugins, w, logger)
-			}
+		hosts := conf.Config.Basic.PluginHosts
+		if err, plugins := fetchPluginList(t, hosts, os, runtime.GOARCH); err != nil {
+			handleError(w, err, "", logger)
 		} else {
-			handleError(w, fmt.Errorf(emsg), "", logger)
-			return
+			jsonResponse(plugins, w, logger)
 		}
 	} else {
 		handleError(w, fmt.Errorf(emsg), "", logger)
diff --git a/internal/topo/neuron/connection.go b/internal/topo/neuron/connection.go
index 12fc28a3..b31b72ea 100644
--- a/internal/topo/neuron/connection.go
+++ b/internal/topo/neuron/connection.go
@@ -38,7 +38,7 @@ var (
 	m               sync.RWMutex
 	connectionCount int
 	sock            mangos.Socket
-	sendTimeout     int
+	sendTimeout     = 100
 )
 
 // createOrGetNeuronConnection creates a new neuron connection or returns an existing one
@@ -105,7 +105,7 @@ func connect(url string) error {
 	//sock.SetOption(mangos.OptionReadQLen, 100)
 	//sock.SetOption(mangos.OptionBestEffort, false)
 	if err = sock.DialOptions(url, map[string]interface{}{
-		mangos.OptionDialAsynch:       false, // will reports error after max reconnect time
+		mangos.OptionDialAsynch:       true, // will not report error and keep connecting
 		mangos.OptionMaxReconnectTime: 5 * time.Second,
 		mangos.OptionReconnectTime:    100 * time.Millisecond,
 	}); err != nil {
diff --git a/internal/topo/neuron/sink.go b/internal/topo/neuron/sink.go
index d0f43c5f..0b5c7b53 100644
--- a/internal/topo/neuron/sink.go
+++ b/internal/topo/neuron/sink.go
@@ -20,6 +20,7 @@ import (
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
+	"github.com/lf-edge/ekuiper/pkg/errorx"
 	"sort"
 )
 
@@ -133,8 +134,7 @@ func (s *sink) SendMapToNeuron(ctx api.StreamContext, el map[string]interface{})
 				t.Value = el[k]
 				err := doPublish(ctx, t)
 				if err != nil {
-					ctx.GetLogger().Error(err)
-					continue
+					return err
 				}
 			}
 		} else {
@@ -143,8 +143,7 @@ func (s *sink) SendMapToNeuron(ctx api.StreamContext, el map[string]interface{})
 				t.Value = v
 				err := doPublish(ctx, t)
 				if err != nil {
-					ctx.GetLogger().Error(err)
-					continue
+					return err
 				}
 			}
 		}
@@ -162,7 +161,7 @@ func (s *sink) SendMapToNeuron(ctx api.StreamContext, el map[string]interface{})
 			}
 			err := doPublish(ctx, t)
 			if err != nil {
-				ctx.GetLogger().Error(err)
+				return err
 			}
 		}
 	}
@@ -176,7 +175,7 @@ func doPublish(ctx api.StreamContext, t *neuronTemplate) error {
 	}
 	err = publish(ctx, r)
 	if err != nil {
-		return fmt.Errorf("Error publish the tag payload %s: %v", t.TagName, err)
+		return fmt.Errorf("%s: Error publish the tag payload %s: %v", errorx.IOErr, t.TagName, err)
 	}
 	ctx.GetLogger().Debugf("Publish %s", r)
 	return nil


Test patch:
diff --git a/internal/topo/neuron/neuron_test.go b/internal/topo/neuron/neuron_test.go
index 17bfec7b..bb5b189b 100644
--- a/internal/topo/neuron/neuron_test.go
+++ b/internal/topo/neuron/neuron_test.go
@@ -105,7 +105,6 @@ func TestMultiSourceSink(t *testing.T) {
 		"groupName": "grp",
 	})
 
-	server, ch := mockNeuron(true, true)
 	wg := sync.WaitGroup{}
 	wg.Add(3)
 	go func() {
@@ -116,7 +115,8 @@ func TestMultiSourceSink(t *testing.T) {
 		mock.TestSourceOpen(s2, exp, t)
 		wg.Done()
 	}()
-
+	// let the server start after the rule to test async dial behavior
+	server, ch := mockNeuron(true, true)
 	data := []interface{}{
 		map[string]interface{}{
 			"temperature": 22,