feat(server): start and stop rules via tags

Signed-off-by: Elia Renzoni <elia.renzoni03@gmail.com>

Author: Elia-Renzoni

internal/server/rest.go

@@ -206,6 +206,8 @@ func createRestServer(ip string, port int, needToken bool) *http.Server {
 	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.MethodPatch, http.MethodDelete)
+	r.HandleFunc("/rules/bulkstart", rulesBulkOperationsHandler).Methods(http.MethodPost)
+	r.HandleFunc("/rules/bulkstop", rulesBulkOperationsHandler).Methods(http.MethodPost)
 	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/rest_test.go

@@ -91,6 +91,8 @@ func (suite *RestTestSuite) SetupTest() {
 	r.HandleFunc("/rules/{name}/topo", getTopoRuleHandler).Methods(http.MethodGet)
 	r.HandleFunc("/rules/{name}/reset_state", ruleStateHandler).Methods(http.MethodPut)
 	r.HandleFunc("/rules/{name}/explain", explainRuleHandler).Methods(http.MethodGet)
+	r.HandleFunc("/rules/bulkstart", rulesBulkOperationsHandler).Methods(http.MethodPost)
+	r.HandleFunc("/rules/bulkstop", rulesBulkOperationsHandler).Methods(http.MethodPost)
 	r.HandleFunc("/rules/{name}/trace/start", enableRuleTraceHandler).Methods(http.MethodPost)
 	r.HandleFunc("/rules/{name}/trace/stop", disableRuleTraceHandler).Methods(http.MethodPost)
 	r.HandleFunc("/rules/validate", validateRuleHandler).Methods(http.MethodPost)
@@ -1024,3 +1026,64 @@ func (suite *RestTestSuite) TestWaitStopRule() {
 	require.True(suite.T(), end.Sub(now) >= 300*time.Millisecond)
 	waitAllRuleStop()
 }
+
+func (suite *RestTestSuite) TestRulesBulkStartAndStop() {
+
+	timestamp := time.Now().UnixNano()
+	mockRules := []string{
+		fmt.Sprintf(`{"id":"r1_%d","sql":"SELECT * FROM demo","actions":[{"log":{}}], "tags": ["mock-tag"], "triggered": true}`, timestamp),
+		fmt.Sprintf(`{"id":"r2_%d","sql":"SELECT color FROM demo","actions":[{"log":{}}], "tags": ["mock-tag"], "triggered": true}`, timestamp),
+		fmt.Sprintf(`{"id":"r3_%d","sql":"SELECT size FROM demo","actions":[{"log":{}}], "tags": ["mock-tag"], "triggered": true}`, timestamp),
+		fmt.Sprintf(`{"id":"r4_%d","sql":"SELECT ts FROM demo","actions":[{"log":{}}], "tags": ["mock-tag"], "triggered": true}`, timestamp),
+		fmt.Sprintf(`{"id":"r5_%d","sql":"SELECT color,size FROM demo","actions":[{"log":{}}], "tags": ["mock-tag"], "triggered": true}`, timestamp),
+		fmt.Sprintf(`{"id":"r6_%d","sql":"SELECT color,ts FROM demo","actions":[{"log":{}}], "tags": ["mock-tag"], "triggered": true}`, timestamp),
+		fmt.Sprintf(`{"id":"r7_%d","sql":"SELECT size,ts FROM demo","actions":[{"log":{}}], "tags": ["mock-tag"], "triggered": true}`, timestamp),
+		fmt.Sprintf(`{"id":"r8_%d","sql":"SELECT color,size,ts FROM demo","actions":[{"log":{}}], "tags": ["mock-tag"], "triggered": true}`, timestamp),
+		fmt.Sprintf(`{"id":"r9_%d","sql":"SELECT * FROM demo WHERE size>0","actions":[{"log":{}}], "tags": ["mock-tag"], "triggered": true}`, timestamp),
+		fmt.Sprintf(`{"id":"r10_%d","sql":"SELECT color FROM demo","actions":[{"log":{}}], "tags": ["mock-tag"], "triggered": true}`, timestamp),
+	}
+
+	// create stream
+	buf := bytes.NewBuffer([]byte(`{"sql":"CREATE STREAM demo (color STRING, size BIGINT, ts BIGINT) WITH (DATASOURCE=\"/data1\", TYPE=\"websocket\", FORMAT=\"json\", KEY=\"ts\")"}`))
+	req, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/streams", buf)
+	w := httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	require.Equal(suite.T(), http.StatusCreated, w.Code)
+
+	// create rules
+	for i, rule := range mockRules {
+		buf := bytes.NewBuffer([]byte(rule))
+		req, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/rules", buf)
+		w := httptest.NewRecorder()
+		suite.r.ServeHTTP(w, req)
+
+		if w.Code != http.StatusCreated {
+			body, _ := io.ReadAll(w.Result().Body)
+			fmt.Printf("Error: %d (index %d): %s\n", i+1, i, string(body))
+		}
+
+		require.Equal(suite.T(), http.StatusCreated, w.Code)
+	}
+
+	// bulk start
+	buf = bytes.NewBuffer([]byte(`{"tags": ["mock-tag"]}`))
+	req, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/rules/bulkstart", buf)
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	if w.Code != http.StatusOK {
+		body, _ := io.ReadAll(w.Result().Body)
+		fmt.Println(string(body))
+	}
+	require.Equal(suite.T(), http.StatusOK, w.Code)
+
+	// bulk stop
+	buf = bytes.NewBuffer([]byte(`{"tags": ["mock-tag"]}`))
+	req, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/rules/bulkstop", buf)
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	if w.Code != http.StatusOK {
+		body, _ := io.ReadAll(w.Result().Body)
+		fmt.Println(string(body))
+	}
+	require.Equal(suite.T(), http.StatusOK, w.Code)
+}

internal/server/rule_tag.go

@@ -16,9 +16,13 @@ package server
 
 import (
 	"encoding/json"
+	"errors"
+	"fmt"
 	"net/http"
+	"path"
 
 	"github.com/gorilla/mux"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/rule"
 )
 
 type RuleTagRequest struct {
@@ -215,3 +219,70 @@ func rulesTagsHandler(w http.ResponseWriter, r *http.Request) {
 	resp := &RuleTagResponse{Rules: res}
 	jsonResponse(resp, w, logger)
 }
+
+func rulesBulkOperationsHandler(w http.ResponseWriter, r *http.Request) {
+	tags := &RuleTagRequest{Tags: []string{}}
+	if err := json.NewDecoder(r.Body).Decode(tags); err != nil {
+		handleError(w, err, "decode body error", logger)
+		return
+	}
+
+	rules, err := ruleProcessor.GetAllRulesJson()
+	if err != nil {
+		handleError(w, err, "", logger)
+		return
+	}
+
+	resultSet := make([]string, 0)
+	for ruleID, ruleJson := range rules {
+		rule, err := ruleProcessor.GetRuleByJsonValidated(ruleID, ruleJson)
+		if err != nil {
+			continue
+		}
+
+		if rule.IsTagsMatch(tags.Tags) {
+			resultSet = append(resultSet, ruleID)
+		}
+	}
+
+	if len(resultSet) == 0 {
+		handleError(w, errors.New(""), "no matching rules", logger)
+		return
+	}
+
+	for _, ruleID := range resultSet {
+		if _, ok := registry.load(ruleID); !ok {
+			ruleJson := rules[ruleID]
+			rr, err := ruleProcessor.GetRuleByJsonValidated(ruleID, ruleJson)
+			if err != nil {
+				continue
+			}
+			rs := rule.NewState(rr, func(id string, b bool) {
+				registry.updateTrigger(id, b)
+			})
+			registry.register(ruleID, rs)
+		}
+	}
+
+	for index, ruleID := range resultSet {
+		var err error
+		if path.Base(r.URL.Path) == "bulkstart" {
+			err = registry.StartRule(ruleID)
+		} else {
+			err = registry.StopRule(ruleID)
+		}
+
+		if err != nil {
+			handleError(
+				w,
+				err,
+				fmt.Sprintf("operation failed, completed %d operations out of %d", index, len(resultSet)),
+				logger,
+			)
+			return
+		}
+	}
+
+	w.WriteHeader(http.StatusOK)
+	w.Write([]byte("operation completed"))
+}