feat(plugin):Add image plugin. (#645)

* feat(plugin):Add image plugin.

* build(CI): fix cross build error (#3)

Co-authored-by: EMqmyd <mayuedong@emx.io>
Co-authored-by: Rory Z <Rory-Z@outlook.com>

.ci/Dockerfile-plugins

@@ -8,7 +8,9 @@ COPY . /go/kuiper
 
 WORKDIR /go/kuiper
 
-RUN apt update && apt install -y pkg-config libczmq-dev jq zip
+RUN set -e -u -x \
+    && apt-get update \
+    && apt-get install -y pkg-config libczmq-dev jq zip --assume-yes apt-utils
 
 RUN set -e -u -x \
     && mkdir -p _plugins/$PLUGIN_TYPE/$PLUGIN_NAME \

.github/workflows/build_packages.yaml

@@ -1,6 +1,8 @@
 name: Build packages
 
 on:
+  push:
+    tags:
   pull_request:
   release:
     types:
@@ -12,15 +14,6 @@ jobs:
 
         steps:
         - uses: actions/checkout@v1
-        - name: install docker
-          run: |
-            sudo apt-get remove docker docker-engine docker.io containerd runc
-            sudo apt-get update
-            sudo apt-get install apt-transport-https ca-certificates curl gnupg-agent software-properties-common
-            curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo apt-key add -
-            sudo add-apt-repository "deb [arch=amd64] https://download.docker.com/linux/ubuntu $(lsb_release -cs) stable"
-            sudo apt-get update
-            sudo apt-get install docker-ce docker-ce-cli containerd.io
         - name: prepare docker
           run: |
             mkdir -p $HOME/.docker
@@ -94,7 +87,7 @@ jobs:
             sudo systemctl restart docker
             docker version
             docker buildx create --use --name mybuild
-            docker run --rm --privileged multiarch/qemu-user-static --reset -p yes
+            docker run --rm --privileged tonistiigi/binfmt --install all
         - name: build docker image
           run: make docker -j4
         - name: buiild debian plugins
@@ -170,7 +163,7 @@ jobs:
             sudo systemctl restart docker
             docker version
             docker buildx create --use --name mybuild
-            docker run --rm --privileged multiarch/qemu-user-static --reset -p yes
+            docker run --rm --privileged tonistiigi/binfmt --install all
         - name: cross build docker images
           if: (matrix.suffix == 'fat') && (github.event_name == 'release')
           run: |
@@ -216,7 +209,7 @@ jobs:
             sudo systemctl restart docker
             docker version
             docker buildx create --use --name mybuild
-            docker run --rm --privileged multiarch/qemu-user-static --reset -p yes
+            docker run --rm --privileged tonistiigi/binfmt --install all
         - name: build docker image
           run: docker build --no-cache -t emqx/kuiper-kubernetes-tool:$(git describe --tags --always) -f deploy/docker/Dockerfile-kubernetes-tool .
         - name: test docker image

Makefile

@@ -89,7 +89,7 @@ real_pkg:
 
 .PHONY:cross_prepare
 cross_prepare:
-	@docker run --rm --privileged multiarch/qemu-user-static --reset -p yes
+	@docker run --rm --privileged tonistiigi/binfmt --install all
 
 .PHONY: cross_build
 cross_build: cross_prepare

docs/en_US/plugins/sinks/image.md

@@ -0,0 +1,52 @@
+# 图像目标（Sink）
+
+目标（Sink）用于将图片保存到指定文件夹中。
+
+## 编译和部署插件
+
+```shell
+# cd $kuiper_src
+# go build --buildmode=plugin -o plugins/sinks/Image.so plugins/sinks/image.go
+# cp plugins/sinks/Image.so $kuiper_install/plugins/sinks
+```
+
+重新启动 Kuiper 服务器以激活插件。
+
+## 属性
+
+| 属性名称 | 是否可选 | 说明                                                         |
+| -------- | -------- | ------------------------------------------------------------ |
+| path     | 否       | 保存图片的文件夹名，例如  `./tmp`。注意：多条 rule 路径不能重复，否则会出现彼此删除的现象。 |
+| format   | 否       | 文件格式，支持 jpeg 和 png。                                 |
+| maxAge   | 是       | 最长文件存储时间(小时)。默认值为72，这表示图片最多保存3天。  |
+| maxCount | 是       | 存储图片的最大数量，默认值是1000，删除时间较早的图片，与`maxAge`是或的关系。 |
+
+## 使用示例
+
+下面示例演示接收图片并将图片保存到文件夹 `/tmp`  中，当图片数量超过1000张时，删除时间较早的图片；当图片保存时长超过72小时时，删除超时的图片。
+
+```json
+{
+  "sql": "SELECT * from demo",
+  "actions": [
+    {
+      "image": {
+        "path": "/tmp",
+        "format": "png",
+        "maxCount":1000,
+        "maxage":72
+      }
+    }
+  ]
+}
+```
+
+## 演示
+
+下面以`zmq`插件为`source`，`image`插件为`sink`，将`zmq`接受到的图片保存在`image`指定的文件夹中。
+
+```shell
+curl http://127.0.0.1:9081/streams -X POST -d '{"sql":"create stream s(image bytea)WITH(DATASOURCE = \"\",FORMAT=\"binary\", TYPE=\"zmq\");"}'
+
+curl http://127.0.0.1:9081/rules -X POST -d '{"id":"r","sql":"SELECT * FROM s","actions":[{"image":{"path":"./tmp","format":"png"}}]}'
+```

docs/zh_CN/plugins/sinks/image.md

@@ -0,0 +1,52 @@
+# 图像目标（Sink）
+
+目标（Sink）用于将图片保存到指定文件夹中。
+
+## 编译和部署插件
+
+```shell
+# cd $kuiper_src
+# go build --buildmode=plugin -o plugins/sinks/Image.so plugins/sinks/image.go
+# cp plugins/sinks/Image.so $kuiper_install/plugins/sinks
+```
+
+重新启动 Kuiper 服务器以激活插件。
+
+## 属性
+
+| 属性名称 | 是否可选 | 说明                                                         |
+| -------- | -------- | ------------------------------------------------------------ |
+| path     | 否       | 保存图片的文件夹名，例如  `./tmp`。注意：多条 rule 路径不能重复，否则会出现彼此删除的现象。 |
+| format   | 否       | 文件格式，支持 jpeg 和 png。                                 |
+| maxAge   | 是       | 最长文件存储时间(小时)。默认值为72，这表示图片最多保存3天。  |
+| maxCount | 是       | 存储图片的最大数量，默认值是1000，删除时间较早的图片，与`maxAge`是或的关系。 |
+
+## 使用示例
+
+下面示例演示接收图片并将图片保存到文件夹 `/tmp`  中，当图片数量超过1000张时，删除时间较早的图片；当图片保存时长超过72小时时，删除超时的图片。
+
+```json
+{
+  "sql": "SELECT * from demo",
+  "actions": [
+    {
+      "image": {
+        "path": "/tmp",
+        "format": "png",
+        "maxCount":1000,
+        "maxage":72
+      }
+    }
+  ]
+}
+```
+
+## 演示
+
+下面以`zmq`插件为`source`，`image`插件为`sink`，将`zmq`接受到的图片保存在`image`指定的文件夹中。
+
+```shell
+curl http://127.0.0.1:9081/streams -X POST -d '{"sql":"create stream s(image bytea)WITH(DATASOURCE = \"\",FORMAT=\"binary\", TYPE=\"zmq\");"}'
+
+curl http://127.0.0.1:9081/rules -X POST -d '{"id":"r","sql":"SELECT * FROM s","actions":[{"image":{"path":"./tmp","format":"png"}}]}'
+```

plugins/sinks/image/image.go

@@ -0,0 +1,187 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"github.com/emqx/kuiper/xstream/api"
+	"image/jpeg"
+	"image/png"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+)
+
+type imageSink struct {
+	path     string
+	format   string
+	maxAge   int
+	maxCount int
+	cancel   context.CancelFunc
+}
+
+func (m *imageSink) Configure(props map[string]interface{}) error {
+	if i, ok := props["format"]; ok {
+		if i, ok := i.(string); ok {
+			if "png" != i && "jpeg" != i {
+				return fmt.Errorf("%s image type is not currently supported", i)
+			}
+			m.format = i
+		}
+	} else {
+		return fmt.Errorf("Field not found format.")
+	}
+
+	if i, ok := props["path"]; ok {
+		if i, ok := i.(string); ok {
+			m.path = i
+		} else {
+			return fmt.Errorf("%s image type is not supported", i)
+		}
+	} else {
+		return fmt.Errorf("Field not found path.")
+	}
+
+	m.maxAge = 72
+	if i, ok := props["maxAge"]; ok {
+		if i, ok := i.(int); ok {
+			m.maxAge = i
+		}
+	}
+	m.maxCount = 1000
+	if i, ok := props["maxCount"]; ok {
+		if i, ok := i.(int); ok {
+			m.maxCount = i
+		}
+	}
+	return nil
+}
+
+func (m *imageSink) Open(ctx api.StreamContext) error {
+	logger := ctx.GetLogger()
+	logger.Debug("Opening image sink")
+
+	if _, err := os.Stat(m.path); os.IsNotExist(err) {
+		if err := os.MkdirAll(m.path, os.ModePerm); nil != err {
+			return fmt.Errorf("fail to open image sink for %v", err)
+		}
+	}
+
+	t := time.NewTicker(time.Duration(3) * time.Minute)
+	exeCtx, cancel := ctx.WithCancel()
+	m.cancel = cancel
+	go func() {
+		defer t.Stop()
+		for {
+			select {
+			case <-t.C:
+				m.delFile(logger)
+			case <-exeCtx.Done():
+				logger.Info("image sink done")
+				return
+			}
+		}
+	}()
+	return nil
+}
+
+func (m *imageSink) delFile(logger api.Logger) error {
+	files, err := ioutil.ReadDir(m.path)
+	if nil != err || 0 == len(files) {
+		return err
+	}
+
+	pos := m.maxCount
+	delTime := time.Now().Add(time.Duration(0-m.maxAge) * time.Hour)
+	for i := 0; i < len(files); i++ {
+		for j := i + 1; j < len(files); j++ {
+			if files[i].ModTime().Before(files[j].ModTime()) {
+				files[i], files[j] = files[j], files[i]
+			}
+		}
+		if files[i].ModTime().Before(delTime) && i < pos {
+			pos = i
+			break
+		}
+	}
+
+	for i := pos; i < len(files); i++ {
+		fname := files[i].Name()
+		if strings.HasSuffix(fname, m.format) {
+			fpath := filepath.Join(m.path, fname)
+			os.Remove(fpath)
+		}
+	}
+	return nil
+}
+
+func (m *imageSink) getSuffix() string {
+	now := time.Now()
+	year, month, day := now.Date()
+	hour, minute, second := now.Clock()
+	nsecond := now.Nanosecond()
+	return fmt.Sprintf(`%d-%d-%d_%d-%d-%d-%d`, year, month, day, hour, minute, second, nsecond)
+}
+
+func (m *imageSink) saveFile(b []byte, fpath string) error {
+	reader := bytes.NewReader(b)
+	fp, err := os.Create(fpath)
+	if nil != err {
+		return err
+	}
+	defer fp.Close()
+	if "png" == m.format {
+		if img, err := png.Decode(reader); nil != err {
+			return err
+		} else if err = png.Encode(fp, img); nil != err {
+			return err
+		}
+	} else if "jpeg" == m.format {
+		if img, err := jpeg.Decode(reader); nil != err {
+			return err
+		} else if err = jpeg.Encode(fp, img, nil); nil != err {
+			return err
+		}
+	}
+	return nil
+}
+
+func (m *imageSink) saveFiles(msg []map[string][]byte) error {
+	for _, images := range msg {
+		for k, v := range images {
+			suffix := m.getSuffix()
+			fname := fmt.Sprintf(`%s%s.%s`, k, suffix, m.format)
+			fpath := filepath.Join(m.path, fname)
+			m.saveFile(v, fpath)
+		}
+	}
+	return nil
+}
+
+func (m *imageSink) Collect(ctx api.StreamContext, item interface{}) error {
+	logger := ctx.GetLogger()
+	if v, ok := item.([]byte); ok {
+		var msg []map[string][]byte
+		if err := json.Unmarshal(v, &msg); nil != err {
+			return fmt.Errorf("The sink only accepts bytea field, other types are not supported.")
+		}
+		return m.saveFiles(msg)
+	} else {
+		logger.Debug("image sink receive non byte data")
+	}
+	return nil
+}
+
+func (m *imageSink) Close(ctx api.StreamContext) error {
+	if m.cancel != nil {
+		m.cancel()
+	}
+	return m.delFile(ctx.GetLogger())
+}
+
+func Image() api.Sink {
+	return &imageSink{}
+}