feat: support rule tags (#3721)

Signed-off-by: Song Gao <disxiaofei@163.com>

Author: Yisaer

docs/en_US/api/restapi/rules.md

@@ -256,3 +256,39 @@ GET http://localhost:9081/rules/usage/cpu
 ```
 
 Get the CPU time used by all rules in the past 30 seconds, in milliseconds.
+
+## Add tags on rules
+
+This API is used to add tags to rules
+
+```shell
+PUT /rules/{id}/tags
+
+{
+  "tags": ["t1","t2"]
+}
+```
+
+## Delete tags on rules
+
+This API is used to delete tags from rules
+
+```shell
+DELETE /rules/{id}/tags
+
+{
+  "keys":["key1","key2"]
+}
+```
+
+## Query rules based on tags
+
+This API is used to query rules containing a given tags and return a list of rule names that meet the conditions
+
+```shell
+GET /rules/tags/match
+
+{
+  "keys":["key1","key2"]
+}
+```

docs/en_US/guide/rules/overview.md

@@ -39,6 +39,7 @@ The table below is a detailed explanation of the row component:
 | graph          | required if sql is not defined   | The json presentation of the rule's DAG(directed acyclic graph)              |
 | options        | true                             | A map of options                                                             |
 | triggerd       | true                             | Whether to start the rule after creation. Default is true.                   |
+| tags           | yes                              | string list, rule tags, used to filter rules                                 |
 
 ## Rule Logic
 

docs/zh_CN/api/restapi/rules.md

@@ -227,3 +227,39 @@ GET http://localhost:9081/rules/usage/cpu
 ```
 
 获取所有规则在过去 30s 内的所使用的 CPU 时间，单位为毫秒
+
+## 添加标签
+
+该 API 用于给规则添加标签
+
+```shell
+PUT /rules/{id}/tags
+
+{
+  "tags": ["t1","t2"]
+}
+```
+
+## 删除标签
+
+该 API 用于给规则删除标签
+
+```shell
+DELETE /rules/{id}/tags
+
+{
+  "tags": ["t1","t2"]
+}
+```
+
+## 根据标签查询规则
+
+该 API 用于根据给定标签查询包含该标签的规则们，返回符合条件的规则名列表
+
+```shell
+GET /rules/tags/match
+
+{
+  "tags": ["t1","t2"]
+}
+```

docs/zh_CN/guide/rules/overview.md

@@ -33,6 +33,7 @@
 | graph    | 如果 sql 未定义，则该属性必须定义   | 规则有向无环图的 JSON 表示                  |
 | options  | 是                     | 选项列表                              |
 | triggerd | 是                     | 布尔值，设置是否创建完规则后立刻运行，默认是 true       |
+| tags     | 是                     | 字符串列表，规则标签，用来筛选规则                 |
 
 ## 规则逻辑
 

fvt/rulestate_test.go

@@ -267,3 +267,66 @@ func (s *RuleStateTestSuite) TestCreateStoppedRule() {
 		s.Equal(http.StatusOK, res.StatusCode)
 	})
 }
+
+func (s *RuleStateTestSuite) TestRuleTags() {
+	s.Run("clean up", func() {
+		client.DeleteStream("simStream1")
+		client.DeleteRule("ruleTags")
+	})
+	s.Run("create rule and attach labels", func() {
+		conf := map[string]any{
+			"interval": "10ms",
+		}
+		resp, err := client.CreateConf("sources/simulator/confKeys/ttt", conf)
+		s.Require().NoError(err)
+		s.Require().Equal(http.StatusOK, resp.StatusCode)
+
+		streamSql := `{"sql": "create stream simStream1() WITH (TYPE=\"simulator\", FORMAT=\"json\", CONF_KEY=\"ttt\", SHARED=\"true\")"}`
+		resp, err = client.CreateStream(streamSql)
+		s.Require().NoError(err)
+		s.T().Log(GetResponseText(resp))
+		s.Require().Equal(http.StatusCreated, resp.StatusCode)
+		ruleJson := `{
+  "id": "ruleTags",
+  "triggered": false,
+  "sql": "SELECT * FROM simStream1",
+  "actions": [
+    {
+      "nop":{}
+    }
+  ],
+  "options": {
+    "sendError": false,
+	"bufferLength": 2
+  }
+}`
+		resp, err = client.CreateRule(ruleJson)
+		s.Require().NoError(err)
+		s.T().Log(GetResponseText(resp))
+		s.Require().Equal(http.StatusCreated, resp.StatusCode)
+
+		resp, err = client.AddRuleTags("ruleTags", []string{"t1", "t2"})
+		s.Require().NoError(err)
+		s.T().Log(GetResponseText(resp))
+		s.Require().Equal(http.StatusOK, resp.StatusCode)
+
+		lists, err := client.GetRulesByTags([]string{"t1", "t2"})
+		s.Require().NoError(err)
+		s.Require().Equal(http.StatusOK, resp.StatusCode)
+		s.Require().Equal([]string{"ruleTags"}, lists)
+
+		resp, err = client.RemoveRuleTags("ruleTags", []string{"t1"})
+		s.Require().NoError(err)
+		s.T().Log(GetResponseText(resp))
+		s.Require().Equal(http.StatusOK, resp.StatusCode)
+
+		lists, err = client.GetRulesByTags([]string{"t1", "t2"})
+		s.Require().NoError(err)
+		s.Require().Equal(http.StatusOK, resp.StatusCode)
+		s.Require().Equal([]string{}, lists)
+	})
+	s.Run("clean up", func() {
+		client.DeleteStream("simStream1")
+		client.DeleteRule("ruleTags")
+	})
+}

fvt/sdk.go

@@ -216,6 +216,47 @@ func (sdk *SDK) BatchRequest(reqs []*server.EachRequest) ([]*server.EachResponse
 	return resps, nil
 }
 
+func (sdk *SDK) AddRuleTags(name string, tags []string) (resp *http.Response, err error) {
+	v, _ := json.Marshal(&server.RuleTagRequest{Tags: tags})
+	url := sdk.baseUrl.JoinPath("rules", name, "tags").String()
+	req, err := http.NewRequest(http.MethodPut, url, bytes.NewBuffer(v))
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+	return sdk.httpClient.Do(req)
+}
+
+func (sdk *SDK) RemoveRuleTags(name string, keys []string) (resp *http.Response, err error) {
+	v, _ := json.Marshal(&server.RuleTagRequest{Tags: keys})
+	req, err := http.NewRequest(http.MethodDelete, sdk.baseUrl.JoinPath("rules", name, "tags").String(), bytes.NewBuffer(v))
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+	return sdk.httpClient.Do(req)
+}
+
+func (sdk *SDK) GetRulesByTags(tags []string) (list []string, err error) {
+	v, _ := json.Marshal(&server.RuleTagRequest{Tags: tags})
+	req, err := http.NewRequest(http.MethodGet, sdk.baseUrl.JoinPath("rules", "tags", "match").String(), bytes.NewBuffer(v))
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+	resp, err := sdk.httpClient.Do(req)
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+	rsp := server.RuleTagResponse{Rules: make([]string, 0)}
+	if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil {
+		fmt.Println(err)
+		return nil, err
+	}
+	return rsp.Rules, nil
+}
+
 func TryAssert(count int, interval time.Duration, tryFunc func() bool) bool {
 	for count > 0 {
 		time.Sleep(interval)

internal/pkg/def/rule.go

@@ -103,6 +103,24 @@ type Rule struct {
 	Graph     *RuleGraph               `json:"graph,omitempty" yaml:"graph,omitempty"`
 	Actions   []map[string]interface{} `json:"actions,omitempty" yaml:"actions,omitempty"`
 	Options   *RuleOption              `json:"options,omitempty" yaml:"options,omitempty"`
+	Tags      []string                 `json:"tags,omitempty" yaml:"tags,omitempty"`
+}
+
+func (r *Rule) IsTagsMatch(tags []string) bool {
+	if len(r.Tags) < len(tags) {
+		return false
+	}
+	mTags := make(map[string]struct{})
+	for _, tag := range r.Tags {
+		mTags[tag] = struct{}{}
+	}
+	for _, tag := range tags {
+		_, ok := mTags[tag]
+		if !ok {
+			return false
+		}
+	}
+	return true
 }
 
 func (r *Rule) IsDurationRule() bool {

internal/server/rest.go

@@ -200,6 +200,8 @@ func createRestServer(ip string, port int, needToken bool) *http.Server {
 	r.HandleFunc("/rules/validate", validateRuleHandler).Methods(http.MethodPost)
 	r.HandleFunc("/rules/{name}/reset_state", ruleStateHandler).Methods(http.MethodPut)
 	r.HandleFunc("/rules/{name}/explain", explainRuleHandler).Methods(http.MethodGet)
+	r.HandleFunc("/rules/tags/match", rulesTagsHandler).Methods(http.MethodGet)
+	r.HandleFunc("/rules/{name}/tags", ruleTagHandler).Methods(http.MethodPut, http.MethodDelete)
 	r.HandleFunc("/ruleset/export", exportHandler).Methods(http.MethodPost)
 	r.HandleFunc("/ruleset/import", importHandler).Methods(http.MethodPost)
 	r.HandleFunc("/configs", configurationUpdateHandler).Methods(http.MethodPatch)

internal/server/rule_manager.go

@@ -49,6 +49,13 @@ type RuleRegistry struct {
 
 //// registry and db level state change functions
 
+func (rr *RuleRegistry) update(key string, ruleJson string, value *rule.State) error {
+	rr.Lock()
+	defer rr.Unlock()
+	rr.internal[key] = value
+	return ruleProcessor.ExecUpsert(key, ruleJson)
+}
+
 // load the entry of a rule by id. It is used to get the current rule state
 // or send command to a running rule
 func (rr *RuleRegistry) load(key string) (value *rule.State, ok bool) {