Merge pull request influxdata#2335 from influxdata/fix/tickerpanic

fix: panic when setting a zero interval for ticker

barrier.go

@@ -59,7 +59,7 @@ func (n *BarrierNode) NewGroup(group edge.GroupInfo, first edge.PointMeta) (edge
 
 func (n *BarrierNode) newBarrier(group edge.GroupInfo, first edge.PointMeta) (edge.ForwardReceiver, func(), error) {
 	switch {
-	case n.b.Idle != 0:
+	case n.b.Idle > 0:
 		idleBarrier := newIdleBarrier(
 			first.Name(),
 			group,
@@ -69,7 +69,7 @@ func (n *BarrierNode) newBarrier(group edge.GroupInfo, first edge.PointMeta) (ed
 			n.b.Delete,
 		)
 		return idleBarrier, idleBarrier.Stop, nil
-	case n.b.Period != 0:
+	case n.b.Period > 0:
 		periodicBarrier := newPeriodicBarrier(
 			first.Name(),
 			group,
@@ -80,7 +80,7 @@ func (n *BarrierNode) newBarrier(group edge.GroupInfo, first edge.PointMeta) (ed
 		)
 		return periodicBarrier, periodicBarrier.Stop, nil
 	default:
-		return nil, nil, errors.New("unreachable code, barrier node should have non-zero idle or non-zero period")
+		return nil, nil, errors.New("unreachable code, barrier node should have positive idle or positive period")
 	}
 }
 

batch.go

@@ -190,14 +190,16 @@ func newQueryNode(et *ExecutingTask, n *pipeline.QueryNode, d NodeDiagnostic) (*
 		return nil, errors.New("must not set both 'every' and 'cron' properties")
 	}
 	switch {
-	case n.Every != 0:
+	case n.Every > 0:
 		bn.ticker = newTimeTicker(n.Every, n.AlignFlag)
 	case n.Cron != "":
 		var err error
 		bn.ticker, err = newCronTicker(n.Cron)
 		if err != nil {
 			return nil, err
 		}
+	case n.Every < 0:
+		return nil, errors.New("'every' duration must must non-negative")
 	default:
 		return nil, errors.New("must define one of 'every' or 'cron'")
 	}

pipeline/json.go

@@ -3,6 +3,7 @@ package pipeline
 import (
 	"bytes"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"sort"
 	"time"
@@ -447,6 +448,9 @@ func unmarshalStats(data []byte, parents []Node, typ TypeOf) (Node, error) {
 	}
 	child := nodeParent.Stats(0)
 	err := json.Unmarshal(data, child)
+	if child.Interval == 0 {
+		return nil, errors.New("zero is an invalid stats interval")
+	}
 	return child, err
 }
 

pipeline/json_test.go

@@ -994,21 +994,39 @@ func Test_unmarshalStats(t *testing.T) {
 			wantErr: true,
 		},
 		{
-			name: "unmarshal stats",
+			name: "should error when interval is zero",
 			args: args{
 				parents: []Node{
 					&node{},
 				},
 				data: []byte(`{
                     "typeOf": "stats",
                     "id": "2",
-                    "interval": "5s",
+                    "interval": "0",
                     "align": true 
                 }`),
 				typ: TypeOf{
 					Type: "stats",
 				},
 			},
+			wantErr: true,
+		},
+		{
+			name: "unmarshal stats",
+			args: args{
+				parents: []Node{
+					&node{},
+				},
+				data: []byte(`{
+		           "typeOf": "stats",
+		           "id": "2",
+		           "interval": "5s",
+		           "align": true
+		       }`),
+				typ: TypeOf{
+					Type: "stats",
+				},
+			},
 			want: &StatsNode{
 				Interval:  5 * time.Second,
 				AlignFlag: true,
@@ -1017,9 +1035,9 @@ func Test_unmarshalStats(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if !tt.wantErr {
-				p := &Pipeline{}
-				p1 := tt.args.parents[0]
+			p := &Pipeline{}
+			p1 := tt.args.parents[0]
+			if p1 != nil {
 				p.addSource(p1)
 			}
 			got, err := unmarshalStats(tt.args.data, tt.args.parents, tt.args.typ)

stats.go

@@ -1,6 +1,7 @@
 package kapacitor
 
 import (
+	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -25,6 +26,11 @@ func newStatsNode(et *ExecutingTask, n *pipeline.StatsNode, d NodeDiagnostic) (*
 	if en == nil {
 		return nil, fmt.Errorf("no node found for %s", n.SourceNode.Name())
 	}
+
+	if n.Interval <= 0 {
+		return nil, errors.New("stats node must have positive interval")
+	}
+
 	sn := &StatsNode{
 		node:    node{Node: n, et: et, diag: d},
 		s:       n,
@@ -41,7 +47,11 @@ func (n *StatsNode) runStats([]byte) error {
 		// Wait till we are roughly aligned with the interval.
 		now := time.Now()
 		next := now.Truncate(n.s.Interval).Add(n.s.Interval)
-		after := time.NewTicker(next.Sub(now))
+		dur := next.Sub(now)
+		if dur <= 0 { // this can happen during a time-changeover like a leap second, or if something is messing about with the system
+			return errors.New("alignment interval should be positive but was not")
+		}
+		after := time.NewTicker(dur)
 		select {
 		case <-after.C:
 			after.Stop()
@@ -53,6 +63,9 @@ func (n *StatsNode) runStats([]byte) error {
 			return err
 		}
 	}
+	if n.s.Interval <= 0 {
+		return errors.New("stats interval should be positive but was not")
+	}
 	ticker := time.NewTicker(n.s.Interval)
 	defer ticker.Stop()
 	for {