chore: add log for rule checker (#2186)

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

internal/conf/conf.go

@@ -28,6 +28,7 @@ import (
 
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
+	"github.com/lf-edge/ekuiper/pkg/schedule"
 )
 
 const ConfFileName = "kuiper.yaml"
@@ -347,6 +348,9 @@ func ValidateRuleOption(option *api.RuleOption) error {
 			errs = errors.Join(errs, errors.New("invalidRestartJitterFactor:restart jitterFactor must between [0, 1)"))
 		}
 	}
+	if err := schedule.ValidateRanges(option.CronDatetimeRange); err != nil {
+		errs = errors.Join(errs, fmt.Errorf("validate cronDatetimeRange failed, err:%v", err))
+	}
 	return errs
 }
 

internal/server/server.go

@@ -302,6 +302,7 @@ func runScheduleRuleChecker(exit <-chan struct{}) {
 
 func handleScheduleRuleState(now time.Time, r *api.Rule, state string) error {
 	toStart, toStop := handleScheduleRule(now, r, state)
+	conf.Log.Debugf("rule %v origin state: %v, going to start:%v, to stop:%v", r.Id, state, toStart, toStop)
 	if toStart {
 		return startRule(r.Id)
 	} else if toStop {

internal/topo/rule/ruleState.go

@@ -147,7 +147,6 @@ func (rs *RuleState) run() {
 				if ctx != nil {
 					conf.Log.Warnf("rule %s is already started", rs.RuleId)
 				} else {
-
 					ctx, cancel = context.WithCancel(context.Background())
 					go rs.runTopo(ctx)
 				}
@@ -369,6 +368,9 @@ func (rs *RuleState) start() error {
 		}
 		rs.triggered = 1
 	}
+	if rs.Rule.IsScheduleRule() || rs.Rule.IsLongRunningScheduleRule() {
+		conf.Log.Debugf("rule %v started", rs.RuleId)
+	}
 	rs.ActionCh <- ActionSignalStart
 	return nil
 }
@@ -377,6 +379,9 @@ func (rs *RuleState) start() error {
 func (rs *RuleState) Stop() error {
 	rs.Lock()
 	defer rs.Unlock()
+	if rs.Rule.IsScheduleRule() || rs.Rule.IsLongRunningScheduleRule() {
+		conf.Log.Debugf("rule %v stopped", rs.RuleId)
+	}
 	rs.stopScheduleRule()
 	return rs.stop()
 }

pkg/schedule/schedule.go

@@ -15,6 +15,7 @@
 package schedule
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/robfig/cron/v3"
@@ -94,3 +95,30 @@ func IsInRunningSchedule(cronExpr string, now time.Time, d time.Duration) (bool,
 	}
 	return false, 0, nil
 }
+
+func ValidateRanges(ranges []api.DatetimeRange) error {
+	if len(ranges) < 1 {
+		return nil
+	}
+	for _, r := range ranges {
+		if err := validateRange(r); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func validateRange(r api.DatetimeRange) error {
+	s, err := cast.InterfaceToTime(r.Begin, layout)
+	if err != nil {
+		return err
+	}
+	e, err := cast.InterfaceToTime(r.End, layout)
+	if err != nil {
+		return err
+	}
+	if s.After(e) {
+		return fmt.Errorf("begin time shouldn't after end time")
+	}
+	return nil
+}

pkg/schedule/schedule_test.go

@@ -15,10 +15,13 @@
 package schedule
 
 import (
+	"errors"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/pkg/api"
 )
 
 func TestIsInTimeRange(t *testing.T) {
@@ -69,3 +72,46 @@ func TestIsInRunningSchedule(t *testing.T) {
 	require.NoError(t, err)
 	require.False(t, isIn)
 }
+
+func TestValidateSchedule(t *testing.T) {
+	tests := []struct {
+		begin string
+		end   string
+		err   error
+	}{
+		{
+			begin: "123",
+			end:   "123",
+			err:   errors.New("Can't parse string as time: 123"),
+		},
+		{
+			begin: layout,
+			end:   "123",
+			err:   errors.New("Can't parse string as time: 123"),
+		},
+		{
+			begin: "2006-01-02 15:04:02",
+			end:   "2006-01-02 15:04:01",
+			err:   errors.New("begin time shouldn't after end time"),
+		},
+		{
+			begin: "2006-01-02 15:04:00",
+			end:   "2006-01-02 15:04:01",
+			err:   nil,
+		},
+	}
+	for _, tc := range tests {
+		rs := []api.DatetimeRange{
+			{
+				Begin: tc.begin,
+				End:   tc.end,
+			},
+		}
+		err := ValidateRanges(rs)
+		if tc.err != nil {
+			require.Equal(t, err, tc.err)
+		} else {
+			require.NoError(t, err)
+		}
+	}
+}