feat: revise rule trace api

internal/pkg/def/rule.go

@@ -39,7 +39,6 @@ type RuleOption struct {
 	Cron               string                   `json:"cron,omitempty" yaml:"cron,omitempty"`
 	Duration           string                   `json:"duration,omitempty" yaml:"duration,omitempty"`
 	CronDatetimeRange  []schedule.DatetimeRange `json:"cronDatetimeRange,omitempty" yaml:"cronDatetimeRange,omitempty"`
-	EnableRuleTracer   bool                     `json:"enableRuleTracer,omitempty" yaml:"enableRuleTracer,omitempty"`
 }
 
 type RestartStrategy struct {

internal/server/rest.go

@@ -169,6 +169,8 @@
 	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("/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)
 	r.HandleFunc("/rules/{name}/reset_state", ruleStateHandler).Methods(http.MethodPut)
 	r.HandleFunc("/rules/{name}/explain", explainRuleHandler).Methods(http.MethodGet)
@@ -790,6 +792,40 @@
 	fmt.Fprintf(w, "Rule %s was restarted", name)
 }
 
+func enableRuleTraceHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	vars := mux.Vars(r)
+	name := vars["name"]
+
+	err := setIsRuleTraceEnabledHandler(name, true)
+	if err != nil {
+		handleError(w, err, "", logger)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+}
+
+func disableRuleTraceHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	vars := mux.Vars(r)
+	name := vars["name"]
+
+	err := setIsRuleTraceEnabledHandler(name, false)
+	if err != nil {
+		handleError(w, err, "", logger)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+}
+
+func setIsRuleTraceEnabledHandler(name string, isEnabled bool) error {
+	rs, ok := registry.load(name)
+	if !ok {
+		return fmt.Errorf("rule %s isn't existed", name)
+	}
+	return rs.SetIsTraceEnabled(isEnabled)
+}
+
 // get topo of a rule
 func getTopoRuleHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()

internal/server/rest_test.go

@@ -89,6 +89,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/{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)
 	r.HandleFunc("/rules/status/all", getAllRuleStatusHandler).Methods(http.MethodGet)
 	r.HandleFunc("/ruleset/export", exportHandler).Methods(http.MethodPost)

internal/server/tracer_test.go

@@ -0,0 +1,47 @@
+// Copyright 2024 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 server
+
+import (
+	"bytes"
+	"net/http"
+	"net/http/httptest"
+
+	"github.com/stretchr/testify/require"
+)
+
+func (suite *RestTestSuite) TestTraceRule() {
+	buf1 := bytes.NewBuffer([]byte(`{"sql":"CREATE stream demo4321() WITH (DATASOURCE=\"0\", TYPE=\"mqtt\")"}`))
+	req1, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/streams", buf1)
+	w1 := httptest.NewRecorder()
+	suite.r.ServeHTTP(w1, req1)
+
+	ruleJson2 := `{"id":"test54321","triggered":false,"sql":"select * from demo4321","actions":[{"log":{}}]}`
+	buf2 := bytes.NewBuffer([]byte(ruleJson2))
+	req2, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/rules", buf2)
+	w2 := httptest.NewRecorder()
+	suite.r.ServeHTTP(w2, req2)
+	require.Equal(suite.T(), http.StatusCreated, w2.Code)
+
+	req2, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/rules/test54321/trace/start", bytes.NewBufferString("any"))
+	w2 = httptest.NewRecorder()
+	suite.r.ServeHTTP(w2, req2)
+	require.Equal(suite.T(), http.StatusOK, w2.Code)
+
+	req2, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/rules/test54321/trace/stop", bytes.NewBufferString("any"))
+	w2 = httptest.NewRecorder()
+	suite.r.ServeHTTP(w2, req2)
+	require.Equal(suite.T(), http.StatusOK, w2.Code)
+}

internal/topo/context/default.go

@@ -21,6 +21,7 @@ import (
 	"regexp"
 	"runtime/pprof"
 	"sync"
+	"sync/atomic"
 	"text/template"
 	"time"
 
@@ -39,12 +40,12 @@ const (
 )
 
 type DefaultContext struct {
-	ruleId           string
-	opId             string
-	instanceId       int
-	ctx              context.Context
-	err              error
-	enableRuleTracer bool
+	ruleId         string
+	opId           string
+	instanceId     int
+	ctx            context.Context
+	err            error
+	isTraceEnabled *atomic.Bool
 	// Only initialized after withMeta set
 	store    api.Store
 	state    *sync.Map
@@ -61,22 +62,28 @@ func RuleBackground(ruleName string) *DefaultContext {
 	ctx := pprof.WithLabels(context.Background(), pprof.Labels("rule", ruleName))
 	pprof.SetGoroutineLabels(ctx)
 	c := &DefaultContext{
-		ctx: ctx,
+		ctx:            ctx,
+		isTraceEnabled: &atomic.Bool{},
 	}
+	c.isTraceEnabled.Store(false)
 	return c
 }
 
 func Background() *DefaultContext {
 	c := &DefaultContext{
-		ctx: context.Background(),
+		ctx:            context.Background(),
+		isTraceEnabled: &atomic.Bool{},
 	}
+	c.isTraceEnabled.Store(false)
 	return c
 }
 
 func WithContext(ctx context.Context) *DefaultContext {
 	c := &DefaultContext{
-		ctx: ctx,
+		ctx:            ctx,
+		isTraceEnabled: &atomic.Bool{},
 	}
+	c.isTraceEnabled.Store(false)
 	return c
 }
 
@@ -197,38 +204,38 @@ func (c *DefaultContext) WithMeta(ruleId string, opId string, store api.Store) a
 		c.GetLogger().Warnf("Initialize context store error for %s: %s", opId, err)
 	}
 	return &DefaultContext{
-		ruleId:           ruleId,
-		opId:             opId,
-		instanceId:       0,
-		ctx:              c.ctx,
-		store:            store,
-		state:            s,
-		tpReg:            sync.Map{},
-		jpReg:            sync.Map{},
-		enableRuleTracer: c.enableRuleTracer,
+		ruleId:         ruleId,
+		opId:           opId,
+		instanceId:     0,
+		ctx:            c.ctx,
+		store:          store,
+		state:          s,
+		tpReg:          sync.Map{},
+		jpReg:          sync.Map{},
+		isTraceEnabled: c.isTraceEnabled,
 	}
 }
 
 func (c *DefaultContext) WithInstance(instanceId int) api.StreamContext {
 	return &DefaultContext{
-		instanceId:       instanceId,
-		ruleId:           c.ruleId,
-		opId:             c.opId,
-		ctx:              c.ctx,
-		state:            c.state,
-		enableRuleTracer: c.enableRuleTracer,
+		instanceId:     instanceId,
+		ruleId:         c.ruleId,
+		opId:           c.opId,
+		ctx:            c.ctx,
+		state:          c.state,
+		isTraceEnabled: c.isTraceEnabled,
 	}
 }
 
 func (c *DefaultContext) WithCancel() (api.StreamContext, context.CancelFunc) {
 	ctx, cancel := context.WithCancel(c.ctx)
 	return &DefaultContext{
-		ruleId:           c.ruleId,
-		opId:             c.opId,
-		instanceId:       c.instanceId,
-		ctx:              ctx,
-		state:            c.state,
-		enableRuleTracer: c.enableRuleTracer,
+		ruleId:         c.ruleId,
+		opId:           c.opId,
+		instanceId:     c.instanceId,
+		ctx:            ctx,
+		state:          c.state,
+		isTraceEnabled: c.isTraceEnabled,
 	}, cancel
 }
 
@@ -291,9 +298,9 @@ func (c *DefaultContext) SaveState(checkpointId int64) error {
 }
 
 func (c *DefaultContext) EnableTracer(enabled bool) {
-	c.enableRuleTracer = enabled
+	c.isTraceEnabled.Store(enabled)
 }
 
 func (c *DefaultContext) IsTraceEnabled() bool {
-	return c.enableRuleTracer
+	return c.isTraceEnabled.Load()
 }

internal/topo/rule/state.go

@@ -550,6 +550,16 @@
 	}
 }
 
+func (s *State) SetIsTraceEnabled(isEnabled bool) error {
+	s.Lock()
+	defer s.Unlock()
+	if s.topology != nil {
+		s.topology.EnableTracer(isEnabled)
+		return nil
+	}
+	return fmt.Errorf("rule %s set trace failed due to rule didn't started", s.Rule.Name)
+}
+
 func (s *State) Delete() (err error) {
 	defer func() {
 		if err != nil {

internal/topo/topo.go

@@ -231,6 +231,10 @@ func (s *Topo) prepareContext() {
 	}
 }
 
+func (s *Topo) EnableTracer(isEnabled bool) {
+	s.ctx.EnableTracer(isEnabled)
+}
+
 func (s *Topo) Open() <-chan error {
 	// if stream has opened, do nothing
 	if s.hasOpened.Load() && !conf.IsTesting {
@@ -242,7 +246,6 @@ func (s *Topo) Open() <-chan error {
 	s.drain = make(chan error, 2)
 	log := s.ctx.GetLogger()
 	log.Info("Opening stream")
-	s.ctx.EnableTracer(s.options.EnableRuleTracer)
 	err := infra.SafeRun(func() error {
 		var err error
 		if s.store, err = state.CreateStore(s.name, s.options.Qos); err != nil {

internal/topo/topotest/rule_test.go

@@ -1244,9 +1244,8 @@ func TestSingleSQL(t *testing.T) {
 	HandleStream(true, streamList, t)
 	options := []*def.RuleOption{
 		{
-			BufferLength:     100,
-			SendError:        true,
-			EnableRuleTracer: true,
+			BufferLength: 100,
+			SendError:    true,
 		},
 		{
 			BufferLength:       100,