Repository: lf-edge/ekuiper
Pull request #1438 (2022-10-17 07:08:26)

feat: support stream/rule import/export


Patch:
diff --git a/cmd/kuiper/main.go b/cmd/kuiper/main.go
index 084ee5de..fb0a9cc0 100644
--- a/cmd/kuiper/main.go
+++ b/cmd/kuiper/main.go
@@ -963,6 +963,64 @@ func main() {
 				},
 			},
 		},
+		{
+			Name:    "import",
+			Aliases: []string{"import"},
+			Usage:   "import ruleset -f ruleset_file",
+			Subcommands: []cli.Command{
+				{
+					Name:  "ruleset",
+					Usage: "\"import ruleset -f ruleset_file",
+					Flags: []cli.Flag{
+						cli.StringFlag{
+							Name:     "file, f",
+							Usage:    "the location of the ruleset json file",
+							FilePath: "/home/ekuiper_ruleset.json",
+						},
+					},
+					Action: func(c *cli.Context) error {
+						sfile := c.String("file")
+						if sfile == "" {
+							fmt.Print("Required ruleset json file to import")
+							return nil
+						}
+						var reply string
+						err = client.Call("Server.Import", sfile, &reply)
+						if err != nil {
+							fmt.Println(err)
+						} else {
+							fmt.Println(reply)
+						}
+						return nil
+					},
+				},
+			},
+		},
+		{
+			Name:    "export",
+			Aliases: []string{"export"},
+			Usage:   "export ruleset $ruleset_file",
+			Subcommands: []cli.Command{
+				{
+					Name:  "ruleset",
+					Usage: "\"export ruleset $ruleset_file",
+					Action: func(c *cli.Context) error {
+						if len(c.Args()) < 1 {
+							fmt.Printf("Require exported file name.\n")
+							return nil
+						}
+						var reply string
+						err = client.Call("Server.Export", c.Args()[0], &reply)
+						if err != nil {
+							fmt.Println(err)
+						} else {
+							fmt.Println(reply)
+						}
+						return nil
+					},
+				},
+			},
+		},
 	}
 
 	app.Name = "Kuiper"
diff --git a/docs/directory.json b/docs/directory.json
index 1c308328..c2d1759a 100644
--- a/docs/directory.json
+++ b/docs/directory.json
@@ -491,12 +491,18 @@
 						{
 							"title": "外部函数管理",
 							"path": "operation/restapi/services"
-						},{
+						},
+						{
 							"title": "模式管理",
 							"path": "operation/restapi/schemas"
-						},{
+						},
+						{
 							"title": "上传文件管理",
 							"path": "operation/restapi/uploads"
+						},
+						{
+							"title": "规则集管理",
+							"path": "operation/restapi/ruleset"
 						}
 					]
 				},
@@ -522,9 +528,14 @@
 						{
 							"title": "插件管理",
 							"path": "operation/cli/plugins"
-						},{
+						},
+						{
 							"title": "模式管理",
 							"path": "operation/cli/schemas"
+						},
+						{
+							"title": "规则集管理",
+							"path": "operation/cli/ruleset"
 						}
 					]
 				},
@@ -1057,12 +1068,18 @@
 						{
 							"title": "External Services",
 							"path": "operation/restapi/services"
-						},{
+						},
+						{
 							"title": "Schemas",
 							"path": "operation/restapi/schemas"
-						},{
+						},
+						{
 							"title": "Upload files",
 							"path": "operation/restapi/uploads"
+						},
+						{
+							"title": "Ruleset",
+							"path": "operation/restapi/ruleset"
 						}
 					]
 				},
@@ -1088,9 +1105,14 @@
 						{
 							"title": "Plugins",
 							"path": "operation/cli/plugins"
-						},{
+						},
+						{
 							"title": "Schemas",
 							"path": "operation/cli/schemas"
+						},
+						{
+							"title": "Ruleset",
+							"path": "operation/cli/ruleset"
 						}
 					]
 				},
diff --git a/docs/en_US/operation/cli/ruleset.md b/docs/en_US/operation/cli/ruleset.md
new file mode 100644
index 00000000..003c6d94
--- /dev/null
+++ b/docs/en_US/operation/cli/ruleset.md
@@ -0,0 +1,37 @@
+# Ruleset Management
+
+The eKuiper rule command line tools allows to import and export all the stream and rule configurations.
+
+## Ruleset Format
+
+The file format for importing and exporting ruleset is JSON, which can contain three parts: `streams`, `tables` and `rules`. Each type holds the the key-value pair of the name and the creation statement. In the following example file, we define a stream and two rules.
+
+```json
+{
+    "streams": {
+        "demo": "CREATE STREAM demo () WITH (DATASOURCE=\"users\", FORMAT=\"JSON\")"
+    },
+    "tables": {},
+    "rules": {
+        "rule1": "{\"id\": \"rule1\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{\"log\": {}}]}",
+        "rule2": "{\"id\": \"rule2\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{  \"log\": {}}]}"
+    }
+}
+```
+
+## Import Ruleset
+
+This command accepts the ruleset and imports it into the system. If a stream or rule in the ruleset already exists, it is not created. The imported rules are started immediately. The command returns text about the number of streams and rules created
+
+
+```shell
+# bin/kuiper import ruleset -f myrules.json
+```
+
+## Export Ruleset
+
+This command exports the ruleset to the specified file. The command returns text about the number of streams and rules exported.
+
+```shell
+# bin/kuiper export ruleset myrules.json
+```
\ No newline at end of file
diff --git a/docs/en_US/operation/restapi/ruleset.md b/docs/en_US/operation/restapi/ruleset.md
new file mode 100644
index 00000000..3ad1f2d2
--- /dev/null
+++ b/docs/en_US/operation/restapi/ruleset.md
@@ -0,0 +1,54 @@
+# Ruleset Management
+
+eKuiper REST api allows to import or export the stream and rule configurations.
+
+## Ruleset Format
+
+The file format for importing and exporting ruleset is JSON, which can contain three parts: `streams`, `tables` and `rules`. Each type holds the the key-value pair of the name and the creation statement. In the following example file, we define a stream and two rules.
+
+```json
+{
+  "streams": {
+    "demo": "CREATE STREAM demo () WITH (DATASOURCE=\"users\", FORMAT=\"JSON\")"
+  },
+  "tables": {},
+  "rules": {
+    "rule1": "{\"id\": \"rule1\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{\"log\": {}}]}",
+    "rule2": "{\"id\": \"rule2\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{  \"log\": {}}]}"
+  }
+}
+```
+
+## Import Ruleset
+
+The API accepts rulesets and imports them into the system. If a stream or rule in the ruleset already exists, it is not created. The API returns text informing the number of streams and rules created. The API supports specifying rulesets by means of text content or file URIs.
+
+Example 1: Import by text content
+
+```shell
+POST http://{{host}}/ruleset/import
+Content-Type: application/json
+
+{
+  "content": "$规则集 json 内容"
+}
+```
+
+Example 2: Import by file URI
+
+```shell
+POST http://{{host}}/ruleset/import
+Content-Type: application/json
+
+{
+  "file": "file:///tmp/a.json"
+}
+```
+
+## Export Ruleset
+
+The export API returns a file to download. 
+
+```shell
+POST http://{{host}}/ruleset/export
+```
\ No newline at end of file
diff --git a/docs/zh_CN/operation/cli/ruleset.md b/docs/zh_CN/operation/cli/ruleset.md
new file mode 100644
index 00000000..2257f55d
--- /dev/null
+++ b/docs/zh_CN/operation/cli/ruleset.md
@@ -0,0 +1,37 @@
+# 规则集管理
+
+eKuiper 命令行工具允许您导入导出当前的所有流和规则配置。
+
+## 规则集格式
+
+导入导出规则集的文件格式为 JSON，其中可包含三个部分：流 `streams`，表 `tables` 和规则 `rules`。每种类型保存名字和创建语句的键值对。在以下示例文件中，我们定义了一个流和两条规则。
+
+```json
+{
+    "streams": {
+        "demo": "CREATE STREAM demo () WITH (DATASOURCE=\"users\", FORMAT=\"JSON\")"
+    },
+    "tables": {},
+    "rules": {
+        "rule1": "{\"id\": \"rule1\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{\"log\": {}}]}",
+        "rule2": "{\"id\": \"rule2\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{  \"log\": {}}]}"
+    }
+}
+```
+
+## 导入规则集
+
+该指令接受规则集并将其导入系统中。若规则集中的流或规则已存在，则不再创建。导入的规则将立刻启动。指令返回文本告知创建的流和规则的数目。
+
+
+```shell
+# bin/kuiper import ruleset -f myrules.json
+```
+
+## 导出规则集
+
+该指令导出规则集到指定的文件中。指令返回文本告知导出的流和规则的数目。
+
+```shell
+# bin/kuiper export ruleset myrules.json
+```
\ No newline at end of file
diff --git a/docs/zh_CN/operation/restapi/ruleset.md b/docs/zh_CN/operation/restapi/ruleset.md
new file mode 100644
index 00000000..953db2e6
--- /dev/null
+++ b/docs/zh_CN/operation/restapi/ruleset.md
@@ -0,0 +1,55 @@
+# 规则集管理
+
+eKuiper REST api 允许您导入导出当前的所有流和规则配置。
+
+## 规则集格式
+
+导入导出规则集的文件格式为 JSON，其中可包含三个部分：流 `streams`，表 `tables` 和规则 `rules`。每种类型保存名字和创建语句的键值对。在以下示例文件中，我们定义了一个流和两条规则。
+
+```json
+{
+    "streams": {
+        "demo": "CREATE STREAM demo () WITH (DATASOURCE=\"users\", FORMAT=\"JSON\")"
+    },
+    "tables": {},
+    "rules": {
+        "rule1": "{\"id\": \"rule1\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{\"log\": {}}]}",
+        "rule2": "{\"id\": \"rule2\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{  \"log\": {}}]}"
+    }
+}
+```
+
+## 导入规则集
+
+该 API 接受规则集并将其导入系统中。若规则集中的流或规则已存在，则不再创建。导入的规则将立刻启动。API 返回文本告知创建的流和规则的数目。 API 支持通过文本内容或者文件 URI 的方式指定规则集。
+
+示例1：通过文本内容导入
+
+
+```shell
+POST http://{{host}}/ruleset/import
+Content-Type: application/json
+
+{
+  "content": "$规则集 json 内容"
+}
+```
+
+示例2：通过文件 URI 导入
+
+```shell
+POST http://{{host}}/ruleset/import
+Content-Type: application/json
+
+{
+  "file": "file:///tmp/a.json"
+}
+```
+
+## 导出规则集
+
+导出 API 返回二进制流，在浏览器使用时，可选择下载保存的文件路径。
+
+```shell
+POST http://{{host}}/ruleset/export
+```
\ No newline at end of file
diff --git a/internal/pkg/httpx/http.go b/internal/pkg/httpx/http.go
index 7b56b9bc..6171953f 100644
--- a/internal/pkg/httpx/http.go
+++ b/internal/pkg/httpx/http.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.
@@ -152,13 +152,14 @@ func IsValidUrl(uri string) bool {
 	return true
 }
 
-func DownloadFile(filepath string, uri string) error {
+// ReadFile Need to close the return reader
+func ReadFile(uri string) (io.ReadCloser, error) {
 	conf.Log.Infof("Start to download file %s\n", uri)
 	u, err := url.ParseRequestURI(uri)
 	if err != nil {
-		return err
+		return nil, err
 	}
-	var src io.Reader
+	var src io.ReadCloser
 	switch u.Scheme {
 	case "file":
 		// deal with windows path
@@ -168,17 +169,16 @@ func DownloadFile(filepath string, uri string) error {
 		conf.Log.Debugf(u.Path)
 		sourceFileStat, err := os.Stat(u.Path)
 		if err != nil {
-			return err
+			return nil, err
 		}
 
 		if !sourceFileStat.Mode().IsRegular() {
-			return fmt.Errorf("%s is not a regular file", u.Path)
+			return nil, fmt.Errorf("%s is not a regular file", u.Path)
 		}
 		srcFile, err := os.Open(u.Path)
 		if err != nil {
-			return err
+			return nil, err
 		}
-		defer srcFile.Close()
 		src = srcFile
 	case "http", "https":
 		// Get the data
@@ -191,16 +191,24 @@ func DownloadFile(filepath string, uri string) error {
 		}
 		resp, err := client.Get(uri)
 		if err != nil {
-			return err
+			return nil, err
 		}
 		if resp.StatusCode != http.StatusOK {
-			return fmt.Errorf("cannot download the file with status: %s", resp.Status)
+			return nil, fmt.Errorf("cannot download the file with status: %s", resp.Status)
 		}
-		defer resp.Body.Close()
 		src = resp.Body
 	default:
-		return fmt.Errorf("unsupported url scheme %s", u.Scheme)
+		return nil, fmt.Errorf("unsupported url scheme %s", u.Scheme)
+	}
+	return src, nil
+}
+
+func DownloadFile(filepath string, uri string) error {
+	src, err := ReadFile(uri)
+	if err != nil {
+		return err
 	}
+	defer src.Close()
 	// Create the file
 	out, err := os.Create(filepath)
 	if err != nil {
diff --git a/internal/pkg/store/redis/redisKv.go b/internal/pkg/store/redis/redisKv.go
index 06baeb5c..f4674a31 100644
--- a/internal/pkg/store/redis/redisKv.go
+++ b/internal/pkg/store/redis/redisKv.go
@@ -22,6 +22,7 @@ import (
 	"encoding/gob"
 	"fmt"
 	"github.com/gomodule/redigo/redis"
+	"github.com/lf-edge/ekuiper/internal/conf"
 	kvEncoding "github.com/lf-edge/ekuiper/internal/pkg/store/encoding"
 	"strings"
 )
@@ -125,6 +126,28 @@ func (kv redisKvStore) Keys() ([]string, error) {
 	return result, nil
 }
 
+func (kv redisKvStore) All() (map[string]string, error) {
+	keys, err := kv.metaKeys()
+	if err != nil {
+		return nil, err
+	}
+	var (
+		value  string
+		result = make(map[string]string)
+	)
+	for _, k := range keys {
+		key := kv.trimPrefix(k)
+		ok, err := kv.Get(key, &value)
+		if err != nil {
+			conf.Log.Errorf("get %s fail during get all in redi: %v", key, err)
+		}
+		if ok {
+			result[key] = value
+		}
+	}
+	return result, nil
+}
+
 func (kv redisKvStore) metaKeys() ([]string, error) {
 	keys := make([]string, 0)
 	err := kv.database.Apply(func(conn redis.Conn) error {
diff --git a/internal/pkg/store/sql/sqlKv.go b/internal/pkg/store/sql/sqlKv.go
index 3109edff..94269781 100644
--- a/internal/pkg/store/sql/sqlKv.go
+++ b/internal/pkg/store/sql/sqlKv.go
@@ -141,6 +141,37 @@ func (kv *sqlKvStore) Keys() ([]string, error) {
 	return keys, err
 }
 
+func (kv *sqlKvStore) All() (all map[string]string, err error) {
+	err = kv.database.Apply(func(db *sql.DB) error {
+		query := fmt.Sprintf("SELECT * FROM '%s'", kv.table)
+		row, e := db.Query(query)
+		if nil != e {
+			return e
+		}
+		defer row.Close()
+		var (
+			key      string
+			valBytes []byte
+			value    string
+		)
+		all = make(map[string]string)
+		for row.Next() {
+			e = row.Scan(&key, &valBytes)
+			if nil != e {
+				return e
+			} else {
+				dec := gob.NewDecoder(bytes.NewBuffer(valBytes))
+				if err := dec.Decode(&value); err != nil {
+					return err
+				}
+				all[key] = value
+			}
+		}
+		return nil
+	})
+	return
+}
+
 func (kv *sqlKvStore) Clean() error {
 	return kv.database.Apply(func(db *sql.DB) error {
 		query := fmt.Sprintf("DELETE FROM '%s'", kv.table)
diff --git a/internal/processor/rule.go b/internal/processor/rule.go
index 038ea5bd..6f6e9a74 100644
--- a/internal/processor/rule.go
+++ b/internal/processor/rule.go
@@ -200,6 +200,10 @@ func (p *RuleProcessor) GetAllRules() ([]string, error) {
 	return p.db.Keys()
 }
 
+func (p *RuleProcessor) GetAllRulesJson() (map[string]string, error) {
+	return p.db.All()
+}
+
 func (p *RuleProcessor) ExecDrop(name string) (string, error) {
 	result := fmt.Sprintf("Rule %s is dropped.", name)
 	var ruleJson string
diff --git a/internal/processor/ruleset.go b/internal/processor/ruleset.go
new file mode 100644
index 00000000..ccef4344
--- /dev/null
+++ b/internal/processor/ruleset.go
@@ -0,0 +1,101 @@
+// Copyright 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package processor
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"io"
+)
+
+type RulesetProcessor struct {
+	r *RuleProcessor
+	s *StreamProcessor
+}
+
+type ruleset struct {
+	Streams map[string]string `json:"streams"`
+	Tables  map[string]string `json:"tables"`
+	Rules   map[string]string `json:"rules"`
+}
+
+func NewRulesetProcessor(r *RuleProcessor, s *StreamProcessor) *RulesetProcessor {
+	return &RulesetProcessor{
+		r: r,
+		s: s,
+	}
+}
+
+func (rs *RulesetProcessor) Export() (io.ReadSeeker, []int, error) {
+	var all ruleset
+	allStreams, err := rs.s.GetAll()
+	if err != nil {
+		return nil, nil, fmt.Errorf("fail to get all streams: %v", err)
+	}
+	all.Streams = allStreams["streams"]
+	all.Tables = allStreams["tables"]
+	rules, err := rs.r.GetAllRulesJson()
+	if err != nil {
+		return nil, nil, fmt.Errorf("fail to get all rules: %v", err)
+	}
+	all.Rules = rules
+	jsonBytes, err := json.Marshal(all)
+	if err != nil {
+		return nil, nil, err
+	}
+	counts := []int{len(all.Streams), len(all.Tables), len(all.Rules)}
+	return bytes.NewReader(jsonBytes), counts, nil
+}
+
+func (rs *RulesetProcessor) Import(content []byte) ([]string, []int, error) {
+	all := &ruleset{}
+	err := json.Unmarshal(content, all)
+	if err != nil {
+		return nil, nil, fmt.Errorf("invalid import file: %v", err)
+	}
+	counts := make([]int, 3)
+	// restore streams
+	for k, v := range all.Streams {
+		_, e := rs.s.ExecStreamSql(v)
+		if e != nil {
+			conf.Log.Errorf("Fail to import stream %s(%s) with error: %v", k, v, e)
+		} else {
+			counts[0]++
+		}
+	}
+	// restore tables
+	for k, v := range all.Tables {
+		_, e := rs.s.ExecStreamSql(v)
+		if e != nil {
+			conf.Log.Errorf("Fail to import table %s(%s) with error: %v", k, v, e)
+		} else {
+			counts[1]++
+		}
+	}
+	var rules []string
+	// restore rules
+	for k, v := range all.Rules {
+		_, e := rs.r.ExecCreate(k, v)
+		if e != nil {
+			conf.Log.Errorf("Fail to import rule %s(%s) with error: %v", k, v, e)
+		} else {
+			rules = append(rules, k)
+			counts[2]++
+		}
+	}
+	return rules, counts, nil
+}
diff --git a/internal/processor/stream.go b/internal/processor/stream.go
index 24ded94f..6e352874 100644
--- a/internal/processor/stream.go
+++ b/internal/processor/stream.go
@@ -398,3 +398,30 @@ func printFieldType(ft ast.FieldType) (result string) {
 	}
 	return
 }
+
+// GetAll return all streams and tables defined to export.
+func (p *StreamProcessor) GetAll() (result map[string]map[string]string, e error) {
+	defs, err := p.db.All()
+	if err != nil {
+		e = err
+		return
+	}
+	var (
+		vs = &xsql.StreamInfo{}
+	)
+	result = map[string]map[string]string{
+		"streams": make(map[string]string),
+		"tables":  make(map[string]string),
+	}
+	for k, v := range defs {
+		if err := json.Unmarshal([]byte(v), vs); err == nil {
+			switch vs.StreamType {
+			case ast.TypeStream:
+				result["streams"][k] = vs.Statement
+			case ast.TypeTable:
+				result["tables"][k] = vs.Statement
+			}
+		}
+	}
+	return
+}
diff --git a/internal/server/rest.go b/internal/server/rest.go
index 391a88b9..e58523ae 100644
--- a/internal/server/rest.go
+++ b/internal/server/rest.go
@@ -15,9 +15,11 @@
 package server
 
 import (
+	"bytes"
 	"encoding/json"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/pkg/httpx"
 	"io"
 	"net/http"
 	"os"
@@ -117,6 +119,8 @@ func createRestServer(ip string, port int, needToken bool) *http.Server {
 	r.HandleFunc("/rules/{name}/stop", stopRuleHandler).Methods(http.MethodPost)
 	r.HandleFunc("/rules/{name}/restart", restartRuleHandler).Methods(http.MethodPost)
 	r.HandleFunc("/rules/{name}/topo", getTopoRuleHandler).Methods(http.MethodGet)
+	r.HandleFunc("/ruleset/export", exportHandler).Methods(http.MethodPost)
+	r.HandleFunc("/ruleset/import", importHandler).Methods(http.MethodPost)
 	r.HandleFunc("/config/uploads", fileUploadHandler).Methods(http.MethodPost, http.MethodGet)
 	r.HandleFunc("/config/uploads/{name}", fileDeleteHandler).Methods(http.MethodDelete)
 
@@ -542,3 +546,67 @@ func getTopoRuleHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set(ContentType, ContentTypeJSON)
 	w.Write([]byte(content))
 }
+
+type rulesetInfo struct {
+	Content  string `json:"content"`
+	FilePath string `json:"file"`
+}
+
+func importHandler(w http.ResponseWriter, r *http.Request) {
+	rsi := &rulesetInfo{}
+	err := json.NewDecoder(r.Body).Decode(rsi)
+	if err != nil {
+		handleError(w, err, "Invalid body: Error decoding json", logger)
+		return
+	}
+	if rsi.Content != "" && rsi.FilePath != "" {
+		handleError(w, nil, "Invalid body: Cannot specify both content and file", logger)
+		return
+	} else if rsi.Content == "" && rsi.FilePath == "" {
+		handleError(w, nil, "Invalid body: must specify content or file", logger)
+		return
+	}
+	content := []byte(rsi.Content)
+	if rsi.FilePath != "" {
+		reader, err := httpx.ReadFile(rsi.FilePath)
+		if err != nil {
+			handleError(w, nil, "Fail to read file", logger)
+			return
+		}
+		defer reader.Close()
+		buf := new(bytes.Buffer)
+		_, err = io.Copy(buf, reader)
+		if err != nil {
+			handleError(w, err, "fail to convert file", logger)
+			return
+		}
+		content = buf.Bytes()
+	}
+	rules, counts, err := rulesetProcessor.Import(content)
+	if err != nil {
+		handleError(w, nil, "Import ruleset error", logger)
+		return
+	}
+	infra.SafeRun(func() error {
+		for _, name := range rules {
+			err := startRule(name)
+			if err != nil {
+				logger.Error(err)
+			}
+		}
+		return nil
+	})
+	w.Write([]byte(fmt.Sprintf("imported %d streams, %d tables and %d rules", counts[0], counts[1], counts[2])))
+}
+
+func exportHandler(w http.ResponseWriter, r *http.Request) {
+	const name = "ekuiper_export.json"
+	exported, _, err := rulesetProcessor.Export()
+	if err != nil {
+		handleError(w, err, "export error", logger)
+		return
+	}
+	w.Header().Set("Content-Type", "application/octet-stream")
+	w.Header().Add("Content-Disposition", "Attachment")
+	http.ServeContent(w, r, name, time.Now(), exported)
+}
diff --git a/internal/server/rpc.go b/internal/server/rpc.go
index 40709e79..12b04d37 100644
--- a/internal/server/rpc.go
+++ b/internal/server/rpc.go
@@ -26,8 +26,10 @@ import (
 	"github.com/lf-edge/ekuiper/internal/pkg/model"
 	"github.com/lf-edge/ekuiper/internal/topo/sink"
 	"github.com/lf-edge/ekuiper/pkg/infra"
+	"io"
 	"net/http"
 	"net/rpc"
+	"os"
 	"strings"
 	"time"
 )
@@ -251,6 +253,52 @@ func (t *Server) DropRule(name string, reply *string) error {
 	return nil
 }
 
+func (t *Server) Import(file string, reply *string) error {
+	f, err := os.Open(file)
+	if err != nil {
+		return fmt.Errorf("fail to read file %s: %v", file, err)
+	}
+	defer f.Close()
+	buf := new(bytes.Buffer)
+	_, err = io.Copy(buf, f)
+	if err != nil {
+		return fmt.Errorf("fail to convert file %s: %v", file, err)
+	}
+	content := buf.Bytes()
+	rules, counts, err := rulesetProcessor.Import(content)
+	if err != nil {
+		return fmt.Errorf("import ruleset error: %v", err)
+	}
+	infra.SafeRun(func() error {
+		for _, name := range rules {
+			err := startRule(name)
+			if err != nil {
+				logger.Error(err)
+			}
+		}
+		return nil
+	})
+	*reply = fmt.Sprintf("imported %d streams, %d tables and %d rules", counts[0], counts[1], counts[2])
+	return nil
+}
+
+func (t *Server) Export(file string, reply *string) error {
+	f, err := os.Create(file)
+	if err != nil {
+		return err
+	}
+	exported, counts, err := rulesetProcessor.Export()
+	if err != nil {
+		return err
+	}
+	_, err = io.Copy(f, exported)
+	if err != nil {
+		return fmt.Errorf("fail to save to file %s:%v", file, err)
+	}
+	*reply = fmt.Sprintf("exported %d streams, %d tables and %d rules", counts[0], counts[1], counts[2])
+	return nil
+}
+
 func marshalDesc(m interface{}) (string, error) {
 	s, err := json.Marshal(m)
 	if err != nil {
diff --git a/internal/server/rule_manager.go b/internal/server/rule_manager.go
index 2c2d140c..cf25a33c 100644
--- a/internal/server/rule_manager.go
+++ b/internal/server/rule_manager.go
@@ -291,24 +291,17 @@ func restartRule(name string) error {
 	return startRule(name)
 }
 
-func recoverRule(name string) string {
-	rule, err := ruleProcessor.GetRuleById(name)
-	if err != nil {
-		return fmt.Sprintf("%v", err)
-	}
-
+func recoverRule(rule *api.Rule) string {
 	if !rule.Triggered {
 		rs := &RuleState{
-			RuleId: name,
+			RuleId: rule.Id,
 		}
-		registry.Store(name, rs)
-		return fmt.Sprintf("Rule %s was stopped.", name)
+		registry.Store(rule.Id, rs)
+		return fmt.Sprintf("Rule %s was stopped.", rule.Id)
 	}
-
-	err = startRule(name)
+	err := startRule(rule.Id)
 	if err != nil {
 		return fmt.Sprintf("%v", err)
 	}
-	return fmt.Sprintf("Rule %s was started.", name)
-
+	return fmt.Sprintf("Rule %s was started.", rule.Id)
 }
diff --git a/internal/server/server.go b/internal/server/server.go
index 724ddfff..9f6d93b1 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -34,11 +34,12 @@ import (
 )
 
 var (
-	logger          = conf.Log
-	startTimeStamp  int64
-	version         = ""
-	ruleProcessor   *processor.RuleProcessor
-	streamProcessor *processor.StreamProcessor
+	logger           = conf.Log
+	startTimeStamp   int64
+	version          = ""
+	ruleProcessor    *processor.RuleProcessor
+	streamProcessor  *processor.StreamProcessor
+	rulesetProcessor *processor.RulesetProcessor
 )
 
 func createPaths() {
@@ -86,6 +87,7 @@ func StartUp(Version, LoadFileType string) {
 
 	ruleProcessor = processor.NewRuleProcessor()
 	streamProcessor = processor.NewStreamProcessor()
+	rulesetProcessor = processor.NewRulesetProcessor(ruleProcessor, streamProcessor)
 
 	// register all extensions
 	for k, v := range components {
@@ -114,7 +116,11 @@ func StartUp(Version, LoadFileType string) {
 	} else {
 		logger.Info("Starting rules")
 		var reply string
-		for _, rule := range rules {
+		for _, name := range rules {
+			rule, err := ruleProcessor.GetRuleById(name)
+			if err != nil {
+				logger.Error(err)
+			}
 			//err = server.StartRule(rule, &reply)
 			reply = recoverRule(rule)
 			if 0 != len(reply) {
diff --git a/pkg/kv/kv.go b/pkg/kv/kv.go
index cc192306..4b2813ea 100644
--- a/pkg/kv/kv.go
+++ b/pkg/kv/kv.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.
@@ -23,6 +23,7 @@ type KeyValue interface {
 	//Must return *common.Error with NOT_FOUND error
 	Delete(key string) error
 	Keys() (keys []string, err error)
+	All() (all map[string]string, err error)
 	Clean() error
 	Drop() error
 }


Test patch:
diff --git a/internal/pkg/store/redis/redisKv_test.go b/internal/pkg/store/redis/redisKv_test.go
index 39544967..ac40ebb9 100644
--- a/internal/pkg/store/redis/redisKv_test.go
+++ b/internal/pkg/store/redis/redisKv_test.go
@@ -51,6 +51,14 @@ func TestRedisKvKeys(t *testing.T) {
 	common.TestKvKeys(length, ks, t)
 }
 
+func TestRedisKvAll(t *testing.T) {
+	ks, db, minRedis := setupRedisKv()
+	defer cleanRedisKv(db, minRedis)
+
+	length := 10
+	common.TestKvAll(length, ks, t)
+}
+
 func setupRedisKv() (kv.KeyValue, *Instance, *miniredis.Miniredis) {
 	minRedis, err := miniredis.Run()
 	if err != nil {
diff --git a/internal/pkg/store/sql/sqlKv_test.go b/internal/pkg/store/sql/sqlKv_test.go
index f7f65dcd..42c50839 100644
--- a/internal/pkg/store/sql/sqlKv_test.go
+++ b/internal/pkg/store/sql/sqlKv_test.go
@@ -54,6 +54,14 @@ func TestSqlKvKeys(t *testing.T) {
 	common.TestKvKeys(length, ks, t)
 }
 
+func TestSqlKvAll(t *testing.T) {
+	ks, db, abs := setupSqlKv()
+	defer cleanSqlKv(db, abs)
+
+	length := 10
+	common.TestKvAll(length, ks, t)
+}
+
 func deleteIfExists(abs string) error {
 	absPath := path.Join(abs, SDbName)
 	if f, _ := os.Stat(absPath); f != nil {
diff --git a/internal/pkg/store/test/common/test.go b/internal/pkg/store/test/common/test.go
index e22c94da..f246baca 100644
--- a/internal/pkg/store/test/common/test.go
+++ b/internal/pkg/store/test/common/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.
@@ -82,3 +82,31 @@ func TestKvKeys(length int, ks kv.KeyValue, t *testing.T) {
 		t.Errorf("Keys do not match expected %s != %s", keys, expected)
 	}
 }
+
+func TestKvAll(length int, ks kv.KeyValue, t *testing.T) {
+
+	expected := make(map[string]string)
+	for i := 0; i < length; i++ {
+		key := fmt.Sprintf("key-%d", i)
+		value := fmt.Sprintf("value-%d", i)
+		if err := ks.Setnx(key, value); err != nil {
+			t.Errorf("It should be set")
+		}
+		expected[key] = value
+	}
+
+	var (
+		all map[string]string
+		err error
+	)
+	if all, err = ks.All(); err != nil {
+		t.Errorf("Failed to get value: %s.", err)
+		return
+	} else if length != len(all) {
+		t.Errorf("expect: %d, got: %d", length, len(all))
+		return
+	}
+	if !reflect.DeepEqual(all, expected) {
+		t.Errorf("All values do not match expected %s != %s", all, expected)
+	}
+}
diff --git a/internal/processor/rule_test.go b/internal/processor/rule_test.go
index 04d0f48a..440e43c4 100644
--- a/internal/processor/rule_test.go
+++ b/internal/processor/rule_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.
@@ -154,3 +154,35 @@ func TestRuleActionParse_Apply(t *testing.T) {
 	}
 
 }
+
+func TestAllRules(t *testing.T) {
+	expected := map[string]string{
+		"rule1": "{\"id\": \"rule1\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{  \"log\": {}}]}",
+		"rule2": "{\"id\": \"rule2\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{  \"log\": {}}]}",
+		"rule3": "{\"id\": \"rule3\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{  \"log\": {}}]}",
+	}
+	sp := NewStreamProcessor()
+	defer sp.db.Clean()
+	sp.ExecStmt(`CREATE STREAM demo () WITH (DATASOURCE="users", FORMAT="JSON")`)
+	p := NewRuleProcessor()
+	p.db.Clean()
+	defer p.db.Clean()
+
+	for k, v := range expected {
+		_, err := p.ExecCreate(k, v)
+		if err != nil {
+			t.Error(err)
+			return
+		}
+		defer p.ExecDrop(k)
+	}
+
+	all, err := p.GetAllRulesJson()
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	if !reflect.DeepEqual(all, expected) {
+		t.Errorf("Expect\t %v\nBut got\t%v", expected, all)
+	}
+}
diff --git a/internal/processor/ruleset_test.go b/internal/processor/ruleset_test.go
new file mode 100644
index 00000000..c1a794d8
--- /dev/null
+++ b/internal/processor/ruleset_test.go
@@ -0,0 +1,108 @@
+// Copyright 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package processor
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/pkg/ast"
+	"io"
+	"reflect"
+	"strings"
+	"testing"
+)
+
+func TestIO(t *testing.T) {
+	expected := `{"streams":{"demo":"CREATE STREAM demo () WITH (DATASOURCE=\"users\", FORMAT=\"JSON\")"},"tables":{},"rules":{"rule1":"{\"id\":\"rule1\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{\"log\": {}}]}","rule2":"{\"id\": \"rule2\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{  \"log\": {}}]}"}}`
+	expectedCounts := []int{1, 0, 2}
+	expectedStreams := []string{"demo"}
+	expectedRules := []string{"rule1", "rule2"}
+	sp := NewStreamProcessor()
+	defer sp.db.Clean()
+	rp := NewRuleProcessor()
+	defer rp.db.Clean()
+	rsp := NewRulesetProcessor(rp, sp)
+
+	names, counts, err := rsp.Import([]byte(expected))
+	if err != nil {
+		t.Errorf("fail to import ruleset: %v", err)
+		return
+	}
+	if !reflect.DeepEqual(names, expectedRules) {
+		t.Errorf("fail to return the imported rules, expect %v but got %v", expectedRules, names)
+	}
+	if !reflect.DeepEqual(counts, expectedCounts) {
+		t.Errorf("fail to return the correct counts, expect %v, but got %v", expectedCounts, counts)
+	}
+
+	streams, err := sp.execShow(ast.TypeStream)
+	if err != nil {
+		t.Errorf("fail to get all streams: %v", err)
+		return
+	}
+	if !reflect.DeepEqual(streams, expectedStreams) {
+		t.Errorf("After import, expect streams %v, but got %v", expectedStreams, streams)
+		return
+	}
+
+	rules, err := rp.GetAllRules()
+	if err != nil {
+		t.Errorf("fail to get all rules: %v", err)
+		return
+	}
+	if !reflect.DeepEqual(rules, expectedRules) {
+		t.Errorf("After import, expect rules %v, but got %v", expectedRules, rules)
+		return
+	}
+
+	exp, exCounts, err := rsp.Export()
+	if err != nil {
+		t.Errorf("fail to export ruleset: %v", err)
+		return
+	}
+	buf := new(strings.Builder)
+	_, err = io.Copy(buf, exp)
+	if err != nil {
+		t.Errorf("fail to convert exported ruleset: %v", err)
+		return
+	}
+	actual := buf.String()
+	if actual != expected {
+		t.Errorf("Expect\t\n %v but got\t\n %v", expected, actual)
+	}
+	if !reflect.DeepEqual(exCounts, expectedCounts) {
+		t.Errorf("fail to return the correct counts, expect %v, but got %v", expectedCounts, exCounts)
+	}
+}
+
+func TestImportError(t *testing.T) {
+	contents := []string{
+		"notjson",
+		`{INvalid"streams":{"demo":"CREATE STREAM demo () WITH (DATASOURCE=\"users\", FORMAT=\"JSON\")"},"tables":{},"rules":{"rule1":"{\"id\":\"rule1\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{\"log\": {}}]}","rule2":"{\"id\": \"rule2\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{  \"log\": {}}]}"}}`,
+	}
+	sp := NewStreamProcessor()
+	defer sp.db.Clean()
+	rp := NewRuleProcessor()
+	defer rp.db.Clean()
+	rsp := NewRulesetProcessor(rp, sp)
+
+	for i, content := range contents {
+		_, _, err := rsp.Import([]byte(content))
+		if err == nil {
+			t.Errorf("%d fail, expect error but pass", i)
+		} else {
+			fmt.Println(err)
+		}
+	}
+}
diff --git a/internal/processor/stream_test.go b/internal/processor/stream_test.go
index 08584b3f..1fda4087 100644
--- a/internal/processor/stream_test.go
+++ b/internal/processor/stream_test.go
@@ -235,3 +235,35 @@ func TestTableList(t *testing.T) {
 		return
 	}
 }
+
+func TestAll(t *testing.T) {
+	expected := map[string]map[string]string{
+		"streams": {
+			"demo":  "create stream demo () WITH (FORMAT=\"JSON\", DATASOURCE=\"demo\", SHARED=\"TRUE\")",
+			"demo1": "create stream demo1 () WITH (FORMAT=\"JSON\", DATASOURCE=\"demo\")",
+			"demo2": "create stream demo2 () WITH (FORMAT=\"JSON\", DATASOURCE=\"demo\", SHARED=\"TRUE\")",
+			"demo3": "create stream demo3 () WITH (FORMAT=\"JSON\", DATASOURCE=\"demo\", SHARED=\"TRUE\")",
+		},
+		"tables": {
+			"tt1": `CREATE TABLE tt1 () WITH (DATASOURCE="users", FORMAT="JSON", KIND="scan")`,
+			"tt3": `CREATE TABLE tt3 () WITH (DATASOURCE="users", TYPE="memory", FORMAT="JSON", KEY="id", KIND="lookup")`,
+		},
+	}
+	p := NewStreamProcessor()
+	p.db.Clean()
+	defer p.db.Clean()
+	for st, m := range expected {
+		for k, v := range m {
+			p.ExecStmt(v)
+			defer p.ExecStmt("Drop " + st + " " + k)
+		}
+	}
+	all, err := p.GetAll()
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	if !reflect.DeepEqual(all, expected) {
+		t.Errorf("Expect\t %v\nBut got\t%v", expected, all)
+	}
+}