Fix watermark (#323)

* don't report watermark if no waiting msg
* report watermark in /status

Author: Ryan-Git

cmd/gravity/main.go

@@ -210,6 +210,7 @@ func statusHandler(server *app.Server, name, hash string) func(http.ResponseWrit
 			Name:       name,
 			ConfigHash: hash,
 			Position:   v,
+			Watermarks: server.Scheduler.Watermarks(),
 			Stage:      state,
 			Version:    utils.Version,
 		}

pkg/core/msg.go

@@ -6,6 +6,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/moiot/gravity/pkg/sliding_window"
+
 	"github.com/juju/errors"
 	"github.com/pingcap/parser/ast"
 	log "github.com/sirupsen/logrus"
@@ -190,11 +192,12 @@ const (
 )
 
 type TaskReportStatus struct {
-	Name       string          `json:"name"`
-	ConfigHash string          `json:"configHash"`
-	Position   string          `json:"position"`
-	Stage      TaskReportStage `json:"stage"`
-	Version    string          `json:"version"`
+	Name       string                              `json:"name"`
+	ConfigHash string                              `json:"configHash"`
+	Position   string                              `json:"position"`
+	Watermarks map[string]sliding_window.Watermark `json:"watermarks"`
+	Stage      TaskReportStage                     `json:"stage"`
+	Version    string                              `json:"version"`
 }
 
 func HashConfig(config string) string {

pkg/core/scheduler.go

@@ -1,9 +1,12 @@
 package core
 
+import "github.com/moiot/gravity/pkg/sliding_window"
+
 type Scheduler interface {
 	MsgSubmitter
 	MsgAcker
 	Healthy() bool
+	Watermarks() map[string]sliding_window.Watermark
 	Start(output Output) error
 	Close()
 }

pkg/schedulers/batch_table_scheduler/batch_table_scheduler.go

@@ -173,6 +173,23 @@ func (scheduler *batchScheduler) Healthy() bool {
 	}
 }
 
+func (scheduler *batchScheduler) Watermarks() (result map[string]sliding_window.Watermark) {
+	if scheduler.cfg.SlidingWindowSize == 0 {
+		return
+	}
+
+	result = make(map[string]sliding_window.Watermark)
+
+	scheduler.windowMutex.Lock()
+	defer scheduler.windowMutex.Unlock()
+
+	for k, w := range scheduler.slidingWindows {
+		result[k] = w.Watermark()
+	}
+
+	return
+}
+
 func (scheduler *batchScheduler) Start(output core.Output) error {
 	if scheduler.cfg.MaxBatchPerWorker > config.TxnBufferLimit-1 {
 		return errors.Errorf("max batch per worker exceed txn buffer limit")

pkg/sliding_window/static_sliding_window.go

@@ -171,19 +171,26 @@ func (w *staticSlidingWindow) reportMetrics() {
 	for {
 		select {
 		case <-ticker.C:
-			watermark := w.Watermark()
-
-			// ProcessTime can be seen as the duration that event are in the queue.
-			seconds := time.Since(watermark.ProcessTime).Seconds()
-			if seconds > ProcessDelayWarningThreshold {
-				log.Warnf("[sliding_window] item not ack after %f seconds. %s", seconds, w.nextItemToCommit)
+			waitingCnt := len(w.waitingItemC)
+
+			if waitingCnt > 0 {
+				watermark := w.Watermark()
+
+				// ProcessTime can be seen as the duration that event are in the queue.
+				seconds := time.Since(watermark.ProcessTime).Seconds()
+				if seconds > ProcessDelayWarningThreshold {
+					log.Warnf("[sliding_window] item not ack after %f seconds. %s", seconds, w.nextItemToCommit)
+				}
+				metrics.End2EndProcessTimeHistogram.WithLabelValues(env.PipelineName).Observe(seconds)
+
+				// EventTime can be seen as the end to end duration of event process time.
+				metrics.End2EndEventTimeHistogram.WithLabelValues(env.PipelineName).Observe(time.Since(watermark.EventTime).Seconds())
+			} else {
+				metrics.End2EndProcessTimeHistogram.WithLabelValues(env.PipelineName).Observe(0)
+				metrics.End2EndEventTimeHistogram.WithLabelValues(env.PipelineName).Observe(0)
 			}
-			metrics.End2EndProcessTimeHistogram.WithLabelValues(env.PipelineName).Observe(seconds)
-
-			// EventTime can be seen as the end to end duration of event process time.
-			metrics.End2EndEventTimeHistogram.WithLabelValues(env.PipelineName).Observe(time.Since(watermark.EventTime).Seconds())
 
-			metrics.QueueLength.WithLabelValues(env.PipelineName, "sliding-window-waiting-chan", w.name).Set(float64(len(w.waitingItemC)))
+			metrics.QueueLength.WithLabelValues(env.PipelineName, "sliding-window-waiting-chan", w.name).Set(float64(waitingCnt))
 			metrics.QueueLength.WithLabelValues(env.PipelineName, "sliding-window-ready-chan", w.name).Set(float64(len(w.readyC)))
 			metrics.QueueLength.WithLabelValues(env.PipelineName, "sliding-window-ready-heap", w.name).Set(float64(atomic.LoadInt64(&w.heapSize)))