Expire groups after X number of zero events

CHANGELOG.md

@@ -168,6 +168,10 @@ batch
 - [#606](https://github.com/influxdata/kapacitor/pull/606): Add Holt-Winters forecasting method.
 - [#605](https://github.com/influxdata/kapacitor/pull/605): BREAKING: StatsNode for batch edge now count the number of points in a batch instead of count batches as a whole.
     This is only breaking if you have a deadman switch configured on a batch edge.
+- [#598](https://github.com/influxdata/kapacitor/pull/598): BREAKING: Change StatsNode to expire groups after X number of points with zero value.
+    This allows for the deadman's switch to expire groups that will never return.
+    This change is only breaking if you relied on the deadman to continually alert on zero values.
+    Now the deadman will alert for only the first X zero values and then stop, effectively setting .stateChangesOnly() for alert node.
 
 ### Bugfixes
 

edge.go

@@ -91,6 +91,14 @@ type edgeStat struct {
 	dims      []string
 }
 
+func (e *Edge) expireGroups(groups []models.GroupID) {
+	e.groupMu.Lock()
+	defer e.groupMu.Unlock()
+	for _, group := range groups {
+		delete(e.groupStats, group)
+	}
+}
+
 // Get a snapshot of the current group statistics for this edge
 func (e *Edge) readGroupStats(f func(group models.GroupID, collected, emitted int64, tags models.Tags, dims []string)) {
 	e.groupMu.RLock()

node.go

@@ -53,6 +53,7 @@ type Node interface {
 	edot(buf *bytes.Buffer, labels bool)
 
 	nodeStatsByGroup() map[models.GroupID]nodeStats
+	expireGroups(groups []models.GroupID)
 
 	collectedCount() int64
 
@@ -64,21 +65,22 @@ type Node interface {
 //implementation of Node
 type node struct {
 	pipeline.Node
-	et         *ExecutingTask
-	parents    []Node
-	children   []Node
-	runF       func(snapshot []byte) error
-	stopF      func()
-	errCh      chan error
-	err        error
-	finishedMu sync.Mutex
-	finished   bool
-	ins        []*Edge
-	outs       []*Edge
-	logger     *log.Logger
-	timer      timer.Timer
-	statsKey   string
-	statMap    *kexpvar.Map
+	et              *ExecutingTask
+	parents         []Node
+	children        []Node
+	runF            func(snapshot []byte) error
+	stopF           func()
+	errCh           chan error
+	err             error
+	finishedMu      sync.Mutex
+	finished        bool
+	ins             []*Edge
+	outs            []*Edge
+	logger          *log.Logger
+	timer           timer.Timer
+	statsKey        string
+	statMap         *kexpvar.Map
+	previousEmitted map[models.GroupID]int64
 }
 
 func (n *node) addParentEdge(e *Edge) {
@@ -103,6 +105,7 @@ func (n *node) init() {
 	n.statMap.Set(statAverageExecTime, avgExecVar)
 	n.timer = n.et.tm.TimingService.NewTimer(avgExecVar)
 	n.errCh = make(chan error, 1)
+	n.previousEmitted = make(map[models.GroupID]int64)
 }
 
 func (n *node) start(snapshot []byte) {
@@ -297,6 +300,16 @@ type nodeStats struct {
 	Fields     models.Fields
 	Tags       models.Tags
 	Dimensions []string
+	IsZero     bool
+}
+
+func (n *node) expireGroups(groups []models.GroupID) {
+	for _, out := range n.outs {
+		out.expireGroups(groups)
+	}
+	for _, group := range groups {
+		delete(n.previousEmitted, group)
+	}
 }
 
 // Return a copy of the current node statistics.
@@ -313,7 +326,9 @@ func (n *node) nodeStatsByGroup() (stats map[models.GroupID]nodeStats) {
 				},
 				Tags:       tags,
 				Dimensions: dims,
+				IsZero:     c == n.previousEmitted[group],
 			}
+			n.previousEmitted[group] = c
 		})
 	}
 	if len(stats) == 0 {

pipeline/stats.go

@@ -1,6 +1,13 @@
 package pipeline
 
-import "time"
+import (
+	"fmt"
+	"time"
+)
+
+const (
+	DefaultExpireCount = int64(1)
+)
 
 // A StatsNode emits internal statistics about the another node at a given interval.
 //
@@ -40,12 +47,25 @@ type StatsNode struct {
 	SourceNode Node
 	// tick:ignore
 	Interval time.Duration
+	// Expire a group after seeing ExpireCount points with a zero value.
+	// An expired group no longer reports stats until a new point
+	// arrives at the source node with that group.
+	// Setting ExpireCount to zero disables expiring groups.
+	ExpireCount int64
 }
 
 func newStatsNode(n Node, interval time.Duration) *StatsNode {
 	return &StatsNode{
-		chainnode:  newBasicChainNode("stats", StreamEdge, StreamEdge),
-		SourceNode: n,
-		Interval:   interval,
+		chainnode:   newBasicChainNode("stats", StreamEdge, StreamEdge),
+		SourceNode:  n,
+		Interval:    interval,
+		ExpireCount: DefaultExpireCount,
+	}
+}
+
+func (n *StatsNode) validate() error {
+	if n.ExpireCount < 0 {
+		return fmt.Errorf("expireCount must be >= 0, got %d", n.ExpireCount)
 	}
+	return nil
 }

stats.go

@@ -17,6 +17,9 @@ type StatsNode struct {
 	closing chan struct{}
 	closed  bool
 	mu      sync.Mutex
+
+	expireCount int
+	zeros       map[models.GroupID]int
 }
 
 // Create a new  FromNode which filters data from a source.
@@ -27,10 +30,12 @@ func newStatsNode(et *ExecutingTask, n *pipeline.StatsNode, l *log.Logger) (*Sta
 		return nil, fmt.Errorf("no node found for %s", n.SourceNode.Name())
 	}
 	sn := &StatsNode{
-		node:    node{Node: n, et: et, logger: l},
-		s:       n,
-		en:      en,
-		closing: make(chan struct{}),
+		node:        node{Node: n, et: et, logger: l},
+		s:           n,
+		en:          en,
+		expireCount: int(n.ExpireCount),
+		closing:     make(chan struct{}),
+		zeros:       make(map[models.GroupID]int),
 	}
 	sn.node.runF = sn.runStats
 	sn.node.stopF = sn.stopStats
@@ -44,6 +49,7 @@ func (s *StatsNode) runStats([]byte) error {
 		Name: "stats",
 		Tags: map[string]string{"node": s.en.Name()},
 	}
+	expiredGroups := make([]models.GroupID, 0)
 	for {
 		select {
 		case <-s.closing:
@@ -57,6 +63,14 @@ func (s *StatsNode) runStats([]byte) error {
 				point.Group = group
 				point.Dimensions = stat.Dimensions
 				point.Tags = stat.Tags
+
+				if stat.IsZero {
+					s.zeros[group]++
+					if s.zeros[group] == s.expireCount {
+						expiredGroups = append(expiredGroups, group)
+						delete(s.zeros, group)
+					}
+				}
 				s.timer.Pause()
 				for _, out := range s.outs {
 					err := out.CollectPoint(point)
@@ -66,6 +80,10 @@ func (s *StatsNode) runStats([]byte) error {
 				}
 				s.timer.Resume()
 			}
+			if len(expiredGroups) > 0 {
+				s.en.expireGroups(expiredGroups)
+				expiredGroups = expiredGroups[0:0]
+			}
 			s.timer.Stop()
 		}
 	}

tick/stateful/evaluation_funcs.go

@@ -1208,6 +1208,24 @@ var evaluationFuncs = map[operationKey]*evaluationFnInfo{
 		},
 		returnType: ast.TDuration,
 	},
+	operationKey{operator: ast.TokenDiv, leftType: ast.TDuration, rightType: ast.TDuration}: {
+		f: func(scope *Scope, executionState ExecutionState, leftNode, rightNode NodeEvaluator) (resultContainer, *ErrSide) {
+			var left time.Duration
+			var right time.Duration
+			var err error
+
+			if left, err = leftNode.EvalDuration(scope, executionState); err != nil {
+				return emptyResultContainer, &ErrSide{error: err, IsLeft: true}
+			}
+
+			if right, err = rightNode.EvalDuration(scope, executionState); err != nil {
+				return emptyResultContainer, &ErrSide{error: err, IsRight: true}
+			}
+
+			return resultContainer{Int64Value: int64(left / right), IsInt64Value: true}, nil
+		},
+		returnType: ast.TInt,
+	},
 
 	// -----------------------------------------
 	//	String concatenation func