Fix bug that reassigned the nodeCardinality

alert.go

@@ -438,12 +438,6 @@ func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode, l *log.Logger) (an *
 		n.History = 2
 	}
 	an.states = make(map[models.GroupID]*alertState)
-	an.nodeCardinality = expvar.NewIntFuncGauge(func() int64 {
-		an.cardinalityMu.RLock()
-		l := len(an.states)
-		an.cardinalityMu.RUnlock()
-		return int64(l)
-	})
 
 	// Configure flapping
 	if n.UseFlapping {
@@ -456,6 +450,12 @@ func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode, l *log.Logger) (an *
 }
 
 func (a *AlertNode) runAlert([]byte) error {
+	a.nodeCardinality.ValueF = func() int64 {
+		a.cardinalityMu.RLock()
+		l := len(a.states)
+		a.cardinalityMu.RUnlock()
+		return int64(l)
+	}
 	// Register delete hook
 	if a.hasAnonTopic() {
 		a.et.tm.registerDeleteHookForTask(a.et.Task.ID, deleteAlertHook(a.anonTopic))

combine.go

@@ -7,7 +7,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/influxdata/kapacitor/expvar"
 	"github.com/influxdata/kapacitor/models"
 	"github.com/influxdata/kapacitor/pipeline"
 	"github.com/influxdata/kapacitor/tick/stateful"
@@ -35,13 +34,6 @@ func newCombineNode(et *ExecutingTask, n *pipeline.CombineNode, l *log.Logger) (
 		combination:        combination{max: n.Max},
 	}
 
-	cn.nodeCardinality = expvar.NewIntFuncGauge(func() int64 {
-		cn.cardinalityMu.RLock()
-		l := len(cn.expressionsByGroup)
-		cn.cardinalityMu.RUnlock()
-		return int64(l)
-	})
-
 	// Create stateful expressions
 	cn.expressions = make([]stateful.Expression, len(n.Lambdas))
 	cn.scopePools = make([]stateful.ScopePool, len(n.Lambdas))
@@ -72,6 +64,13 @@ func (t timeList) Less(i, j int) bool { return t[i].Before(t[j]) }
 func (t timeList) Swap(i, j int)      { t[i], t[j] = t[j], t[i] }
 
 func (n *CombineNode) runCombine([]byte) error {
+	n.nodeCardinality.ValueF = func() int64 {
+		n.cardinalityMu.RLock()
+		l := len(n.expressionsByGroup)
+		n.cardinalityMu.RUnlock()
+		return int64(l)
+	}
+
 	switch n.Wants() {
 	case pipeline.StreamEdge:
 		buffers := make(map[models.GroupID]*buffer)

derivative.go

@@ -5,7 +5,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/influxdata/kapacitor/expvar"
 	"github.com/influxdata/kapacitor/models"
 	"github.com/influxdata/kapacitor/pipeline"
 )
@@ -31,12 +30,12 @@ func (d *DerivativeNode) runDerivative([]byte) error {
 	case pipeline.StreamEdge:
 		var mu sync.RWMutex
 		previous := make(map[models.GroupID]models.Point)
-		d.nodeCardinality = expvar.NewIntFuncGauge(func() int64 {
+		d.nodeCardinality.ValueF = func() int64 {
 			mu.RLock()
 			l := len(previous)
 			mu.RUnlock()
 			return int64(l)
-		})
+		}
 		for p, ok := d.ins[0].NextPoint(); ok; p, ok = d.ins[0].NextPoint() {
 			d.timer.Start()
 			mu.RLock()

eval.go

@@ -38,12 +38,6 @@ func newEvalNode(et *ExecutingTask, n *pipeline.EvalNode, l *log.Logger) (*EvalN
 		e:                  n,
 		expressionsByGroup: make(map[models.GroupID][]stateful.Expression),
 	}
-	en.nodeCardinality = expvar.NewIntFuncGauge(func() int64 {
-		en.cardinalityMu.RLock()
-		l := len(en.expressionsByGroup)
-		en.cardinalityMu.RUnlock()
-		return int64(l)
-	})
 
 	// Create stateful expressions
 	en.expressions = make([]stateful.Expression, len(n.Lambdas))
@@ -75,6 +69,12 @@ func newEvalNode(et *ExecutingTask, n *pipeline.EvalNode, l *log.Logger) (*EvalN
 }
 
 func (e *EvalNode) runEval(snapshot []byte) error {
+	e.nodeCardinality.ValueF = func() int64 {
+		e.cardinalityMu.RLock()
+		l := len(e.expressionsByGroup)
+		e.cardinalityMu.RUnlock()
+		return int64(l)
+	}
 	switch e.Provides() {
 	case pipeline.StreamEdge:
 		var err error

expvar/expvar.go

@@ -59,6 +59,9 @@ func (v *IntFuncGauge) Add(delta int64) {}
 func (v *IntFuncGauge) Set(value int64) {}
 
 func (v *IntFuncGauge) IntValue() int64 {
+	if v == nil || v.ValueF == nil {
+		return 0
+	}
 	return v.ValueF()
 }
 

flatten.go

@@ -7,7 +7,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/influxdata/kapacitor/expvar"
 	"github.com/influxdata/kapacitor/models"
 	"github.com/influxdata/kapacitor/pipeline"
 )
@@ -54,12 +53,12 @@ func (n *FlattenNode) runFlatten([]byte) error {
 	switch n.Wants() {
 	case pipeline.StreamEdge:
 		flattenBuffers := make(map[models.GroupID]*flattenStreamBuffer)
-		n.nodeCardinality = expvar.NewIntFuncGauge(func() int64 {
+		n.nodeCardinality.ValueF = func() int64 {
 			mu.RLock()
 			l := len(flattenBuffers)
 			mu.RUnlock()
 			return int64(l)
-		})
+		}
 
 		for p, ok := n.ins[0].NextPoint(); ok; p, ok = n.ins[0].NextPoint() {
 			n.timer.Start()
@@ -117,12 +116,12 @@ func (n *FlattenNode) runFlatten([]byte) error {
 		}
 	case pipeline.BatchEdge:
 		allBuffers := make(map[models.GroupID]*flattenBatchBuffer)
-		n.nodeCardinality = expvar.NewIntFuncGauge(func() int64 {
+		n.nodeCardinality.ValueF = func() int64 {
 			mu.RLock()
 			l := len(allBuffers)
 			mu.RUnlock()
 			return int64(l)
-		})
+		}
 		for b, ok := n.ins[0].NextBatch(); ok; b, ok = n.ins[0].NextBatch() {
 			n.timer.Start()
 			t := b.TMax.Round(n.f.Tolerance)

http_out.go

@@ -7,7 +7,6 @@ import (
 	"path"
 	"sync"
 
-	"github.com/influxdata/kapacitor/expvar"
 	"github.com/influxdata/kapacitor/models"
 	"github.com/influxdata/kapacitor/pipeline"
 	"github.com/influxdata/kapacitor/services/httpd"
@@ -31,12 +30,6 @@ func newHTTPOutNode(et *ExecutingTask, n *pipeline.HTTPOutNode, l *log.Logger) (
 		groupSeriesIdx: make(map[models.GroupID]int),
 		result:         new(models.Result),
 	}
-	hn.nodeCardinality = expvar.NewIntFuncGauge(func() int64 {
-		hn.mu.RLock()
-		l := len(hn.groupSeriesIdx)
-		hn.mu.RUnlock()
-		return int64(l)
-	})
 	et.registerOutput(hn.c.Endpoint, hn)
 	hn.node.runF = hn.runOut
 	hn.node.stopF = hn.stopOut
@@ -48,6 +41,13 @@ func (h *HTTPOutNode) Endpoint() string {
 }
 
 func (h *HTTPOutNode) runOut([]byte) error {
+	h.nodeCardinality.ValueF = func() int64 {
+		h.mu.RLock()
+		l := len(h.groupSeriesIdx)
+		h.mu.RUnlock()
+		return int64(l)
+	}
+
 	hndl := func(w http.ResponseWriter, req *http.Request) {
 		h.mu.RLock()
 		defer h.mu.RUnlock()

influxql.go

@@ -6,7 +6,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/influxdata/kapacitor/expvar"
 	"github.com/influxdata/kapacitor/models"
 	"github.com/influxdata/kapacitor/pipeline"
 	"github.com/pkg/errors"
@@ -74,12 +73,12 @@ func (c *baseReduceContext) Time() time.Time {
 func (n *InfluxQLNode) runStreamInfluxQL() error {
 	var mu sync.RWMutex
 	contexts := make(map[models.GroupID]reduceContext)
-	n.nodeCardinality = expvar.NewIntFuncGauge(func() int64 {
+	n.nodeCardinality.ValueF = func() int64 {
 		mu.RLock()
 		l := len(contexts)
 		mu.RUnlock()
 		return int64(l)
-	})
+	}
 
 	for p, ok := n.ins[0].NextPoint(); ok; {
 		n.timer.Start()

join.go

@@ -68,12 +68,12 @@ func newJoinNode(et *ExecutingTask, n *pipeline.JoinNode, l *log.Logger) (*JoinN
 
 func (j *JoinNode) runJoin([]byte) error {
 	j.groups = make(map[models.GroupID]*group)
-	j.nodeCardinality = expvar.NewIntFuncGauge(func() int64 {
+	j.nodeCardinality.ValueF = func() int64 {
 		j.cardinalityMu.RLock()
 		l := len(j.groups)
 		j.cardinalityMu.RUnlock()
 		return int64(l)
-	})
+	}
 
 	groupErrs := make(chan error, 1)
 	done := make(chan struct{}, len(j.ins))

k8s_autoscale.go

@@ -63,18 +63,19 @@ func newK8sAutoscaleNode(et *ExecutingTask, n *pipeline.K8sAutoscaleNode, l *log
 	// Initialize the replicas lambda expression scope pool
 	if n.Replicas != nil {
 		kn.replicasExprs = make(map[models.GroupID]stateful.Expression)
-		kn.nodeCardinality = expvar.NewIntFuncGauge(func() int64 {
-			kn.cardinalityMu.RLock()
-			l := len(kn.replicasExprs)
-			kn.cardinalityMu.RUnlock()
-			return int64(l)
-		})
 		kn.replicasScopePool = stateful.NewScopePool(stateful.FindReferenceVariables(n.Replicas.Expression))
 	}
 	return kn, nil
 }
 
 func (k *K8sAutoscaleNode) runAutoscale([]byte) error {
+	k.nodeCardinality.ValueF = func() int64 {
+		k.cardinalityMu.RLock()
+		l := len(k.replicasExprs)
+		k.cardinalityMu.RUnlock()
+		return int64(l)
+	}
+
 	k.increaseCount = &expvar.Int{}
 	k.decreaseCount = &expvar.Int{}
 	k.cooldownDropsCount = &expvar.Int{}

node.go

@@ -19,7 +19,8 @@ import (
 )
 
 const (
-	statErrorCount        = "errors"
+	statErrorCount = "errors"
+	// TODO: Change to working_cardinality
 	statsCardinalityGauge = "cardinality"
 	statAverageExecTime   = "avg_exec_time_ns"
 )
@@ -111,7 +112,7 @@ func (n *node) init() {
 	n.statMap.Set(statAverageExecTime, avgExecVar)
 	n.nodeErrors = &kexpvar.Int{}
 	n.statMap.Set(statErrorCount, n.nodeErrors)
-	n.nodeCardinality = kexpvar.NewIntFuncGauge(func() int64 { return 0 })
+	n.nodeCardinality = kexpvar.NewIntFuncGauge(nil)
 	n.statMap.Set(statsCardinalityGauge, n.nodeCardinality)
 	n.timer = n.et.tm.TimingService.NewTimer(avgExecVar)
 	n.errCh = make(chan error, 1)

sample.go

@@ -6,7 +6,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/influxdata/kapacitor/expvar"
 	"github.com/influxdata/kapacitor/models"
 	"github.com/influxdata/kapacitor/pipeline"
 )
@@ -29,12 +28,6 @@ func newSampleNode(et *ExecutingTask, n *pipeline.SampleNode, l *log.Logger) (*S
 		counts:   make(map[models.GroupID]int64),
 		duration: n.Duration,
 	}
-	sn.nodeCardinality = expvar.NewIntFuncGauge(func() int64 {
-		sn.cardinalityMu.RLock()
-		l := len(sn.counts)
-		sn.cardinalityMu.RUnlock()
-		return int64(l)
-	})
 	sn.node.runF = sn.runSample
 	if n.Duration == 0 && n.N == 0 {
 		return nil, errors.New("invalid sample rate: must be positive integer or duration")
@@ -43,6 +36,13 @@ func newSampleNode(et *ExecutingTask, n *pipeline.SampleNode, l *log.Logger) (*S
 }
 
 func (s *SampleNode) runSample([]byte) error {
+	s.nodeCardinality.ValueF = func() int64 {
+		s.cardinalityMu.RLock()
+		l := len(s.counts)
+		s.cardinalityMu.RUnlock()
+		return int64(l)
+	}
+
 	switch s.Wants() {
 	case pipeline.StreamEdge:
 		for p, ok := s.ins[0].NextPoint(); ok; p, ok = s.ins[0].NextPoint() {

where.go

@@ -6,7 +6,6 @@ import (
 	"log"
 	"sync"
 
-	"github.com/influxdata/kapacitor/expvar"
 	"github.com/influxdata/kapacitor/models"
 	"github.com/influxdata/kapacitor/pipeline"
 	"github.com/influxdata/kapacitor/tick/stateful"
@@ -38,12 +37,12 @@ func newWhereNode(et *ExecutingTask, n *pipeline.WhereNode, l *log.Logger) (wn *
 
 func (w *WhereNode) runWhere(snapshot []byte) error {
 	var mu sync.RWMutex
-	w.nodeCardinality = expvar.NewIntFuncGauge(func() int64 {
+	w.nodeCardinality.ValueF = func() int64 {
 		mu.RLock()
 		l := len(w.expressions)
 		mu.RUnlock()
 		return int64(l)
-	})
+	}
 
 	switch w.Wants() {
 	case pipeline.StreamEdge:

window.go

@@ -7,7 +7,6 @@ import (
 	"sync"
 	"time"
 
-	kexpvar "github.com/influxdata/kapacitor/expvar"
 	"github.com/influxdata/kapacitor/models"
 	"github.com/influxdata/kapacitor/pipeline"
 )
@@ -34,12 +33,13 @@ type window interface {
 func (w *WindowNode) runWindow([]byte) error {
 	var mu sync.RWMutex
 	windows := make(map[models.GroupID]window)
-	w.nodeCardinality = kexpvar.NewIntFuncGauge(func() int64 {
+	w.nodeCardinality.ValueF = func() int64 {
 		mu.RLock()
 		l := len(windows)
 		mu.RUnlock()
 		return int64(l)
-	})
+	}
+
 	// Loops through points windowing by group
 	for p, ok := w.ins[0].NextPoint(); ok; p, ok = w.ins[0].NextPoint() {
 		w.timer.Start()