feat: support schedule rule status timestamp

internal/server/rule_manager.go

@@ -15,7 +15,6 @@
 package server
 
 import (
-	"bytes"
 	"encoding/json"
 	"fmt"
 	"sort"
@@ -28,7 +27,6 @@ import (
 	"github.com/lf-edge/ekuiper/internal/topo/rule"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
-	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
 	"github.com/lf-edge/ekuiper/pkg/infra"
 )
@@ -248,30 +246,26 @@ func getRuleStatus(name string) (string, error) {
 		if err != nil {
 			return "", err
 		}
+		m := map[string]interface{}{}
 		if result == "Running" {
+			m["status"] = "running"
 			keys, values := (*rs.Topology).GetMetrics()
-			metrics := "{"
-			metrics += `"status": "running",`
 			for i, key := range keys {
-				value := values[i]
-				switch value.(type) {
-				case string:
-					metrics += fmt.Sprintf("\"%s\":%q,", key, value)
-				default:
-					metrics += fmt.Sprintf("\"%s\":%v,", key, value)
-				}
-			}
-			metrics = metrics[:len(metrics)-1] + "}"
-			dst := &bytes.Buffer{}
-			if err = json.Indent(dst, cast.StringToBytes(metrics), "", "  "); err != nil {
-				result = metrics
-			} else {
-				result = dst.String()
+				m[key] = values[i]
 			}
 		} else {
-			result = fmt.Sprintf(`{"status": "stopped", "message": "%s"}`, result)
+			m["status"] = "stopped"
+			m["message"] = result
+		}
+		lastStart, lastStop, nextStart := rs.GetScheduleTimestamp()
+		m["lastStartTimestamp"] = lastStart
+		m["lastStopTimestamp"] = lastStop
+		m["nextStartTimestamp"] = nextStart
+		content, err := json.Marshal(m)
+		if err != nil {
+			return "", err
 		}
-		return result, nil
+		return string(content), nil
 	} else {
 		return "", errorx.NewWithCode(errorx.NOT_FOUND, fmt.Sprintf("Rule %s is not found", name))
 	}

internal/topo/rule/ruleState.go

@@ -97,6 +97,9 @@ type RuleState struct {
 	topoGraph *api.PrintableTopo
 	sync.RWMutex
 	cronState cronStateCtx
+
+	lastStartTimestamp int64
+	lastStopTimestamp  int64
 }
 
 // NewRuleState Create and initialize a rule state.
@@ -396,6 +399,7 @@ func (rs *RuleState) start() error {
 	if rs.Rule.IsScheduleRule() || rs.Rule.IsLongRunningScheduleRule() {
 		conf.Log.Debugf("rule %v started", rs.RuleId)
 	}
+	rs.lastStartTimestamp = time.Now().UnixMilli()
 	rs.ActionCh <- ActionSignalStart
 	return nil
 }
@@ -430,6 +434,7 @@ func (rs *RuleState) stop() error {
 	if rs.Topology != nil {
 		rs.Topology.Cancel()
 	}
+	rs.lastStopTimestamp = time.Now().UnixMilli()
 	rs.ActionCh <- ActionSignalStop
 	return nil
 }
@@ -558,3 +563,23 @@ func (rs *RuleState) isInAllowedTimeRange(now time.Time) (bool, error) {
 	}
 	return allowed, nil
 }
+
+func (rs *RuleState) GetNextScheduleStartTime() int64 {
+	if rs.Rule.IsScheduleRule() && len(rs.Rule.Options.Cron) > 0 {
+		isIn, err := schedule.IsInScheduleRanges(time.Now(), rs.Rule.Options.CronDatetimeRange)
+		if err == nil && isIn {
+			s, err := cron.ParseStandard(rs.Rule.Options.Cron)
+			if err == nil {
+				return s.Next(time.Now()).UnixMilli()
+			}
+		}
+	}
+	return 0
+}
+
+func (rs *RuleState) GetScheduleTimestamp() (int64, int64, int64) {
+	nextStartTimestamp := rs.GetNextScheduleStartTime()
+	rs.Lock()
+	defer rs.Unlock()
+	return rs.lastStartTimestamp, rs.lastStopTimestamp, nextStartTimestamp
+}