Group By Fields (influxdata#731)

Author: Nathaniel Cook

CHANGELOG.md

@@ -4,9 +4,57 @@
 
 ### Release Notes
 
+#### Group By Fields
+
+Kapacitor now supports grouping by fields.
+First convert a field into a tag using the EvalNode.
+Then group by the new tag.
+
+Example:
+
+```go
+stream
+    |from()
+        .measurement('alerts')
+    // Convert field 'level' to tag.
+    |eval(lambda: string("level"))
+        .as('level')
+        .tags('level')
+    // Group by new tag 'level'.
+    |groupBy('alert', 'level')
+    |...
+```
+
+Note the field `level` is now removed from the point since `.keep` was not used.
+See the [docs](https://docs.influxdata.com/kapacitor/v1.0/nodes/eval_node/#tags) for more details on how `.tags` works.
+
+
+#### Delete Fields or Tags
+
+In companion with being able to create new tags, you can now delete tags or fields.
+
+
+Example:
+
+```go
+stream
+    |from()
+        .measurement('alerts')
+    |delete()
+        // Remove the field `extra` and tag `uuid` from all points.
+        .field('extra')
+        .tag('uuid')
+    |...
+```
+
+
+
 ### Features
 
 - [#702](https://github.com/influxdata/kapacitor/pull/702): Add plumbing for authentication backends.
+- [#624](https://github.com/influxdata/kapacitor/issue/624): BREAKING: Add ability to GroupBy fields. First use EvalNode to create a tag from a field and then group by the new tag.
+    Also allows for grouping by measurement.
+    The breaking change is that the group ID format has changed to allow for the measurement name.
 
 ### Bugfixes
 

alert.go

@@ -385,6 +385,7 @@ func (a *AlertNode) runAlert([]byte) error {
 				batch := models.Batch{
 					Name:   p.Name,
 					Group:  p.Group,
+					ByName: p.Dimensions.ByName,
 					Tags:   p.Tags,
 					Points: []models.BatchPoint{models.BatchPointFromPoint(p)},
 				}

batch.go

@@ -95,17 +95,19 @@ func (s *BatchNode) Abort() {
 }
 
 type BatchQueries struct {
-	Queries []string
-	Cluster string
+	Queries            []string
+	Cluster            string
+	GroupByMeasurement bool
 }
 
 func (s *BatchNode) Queries(start, stop time.Time) []BatchQueries {
 	queries := make([]BatchQueries, len(s.children))
 	for i, b := range s.children {
 		qn := b.(*QueryNode)
 		queries[i] = BatchQueries{
-			Queries: qn.Queries(start, stop),
-			Cluster: qn.Cluster(),
+			Queries:            qn.Queries(start, stop),
+			Cluster:            qn.Cluster(),
+			GroupByMeasurement: qn.GroupByMeasurement(),
 		}
 	}
 	return queries
@@ -136,6 +138,7 @@ type QueryNode struct {
 	connectErrors  *expvar.Int
 	batchesQueried *expvar.Int
 	pointsQueried  *expvar.Int
+	byName         bool
 }
 
 func newQueryNode(et *ExecutingTask, n *pipeline.QueryNode, l *log.Logger) (*QueryNode, error) {
@@ -144,6 +147,7 @@ func newQueryNode(et *ExecutingTask, n *pipeline.QueryNode, l *log.Logger) (*Que
 		b:        n,
 		closing:  make(chan struct{}),
 		aborting: make(chan struct{}),
+		byName:   n.GroupByMeasurementFlag,
 	}
 	bn.node.runF = bn.runBatch
 	bn.node.stopF = bn.stopBatch
@@ -196,6 +200,10 @@ func newQueryNode(et *ExecutingTask, n *pipeline.QueryNode, l *log.Logger) (*Que
 	return bn, nil
 }
 
+func (b *QueryNode) GroupByMeasurement() bool {
+	return b.byName
+}
+
 // Return list of databases and retention policies
 // the batcher will query.
 func (b *QueryNode) DBRPs() ([]DBRP, error) {
@@ -319,7 +327,7 @@ func (b *QueryNode) doQuery() error {
 
 			// Collect batches
 			for _, res := range resp.Results {
-				batches, err := models.ResultToBatches(res)
+				batches, err := models.ResultToBatches(res, b.byName)
 				if err != nil {
 					b.logger.Println("E! failed to understand query result:", err)
 					b.queryErrors.Add(1)

delete.go

@@ -0,0 +1,97 @@
+package kapacitor
+
+import (
+	"log"
+
+	"github.com/influxdata/kapacitor/expvar"
+	"github.com/influxdata/kapacitor/models"
+	"github.com/influxdata/kapacitor/pipeline"
+)
+
+const (
+	statsFieldsDeleted = "fields_deleted"
+	statsTagsDeleted   = "tags_deleted"
+)
+
+type DeleteNode struct {
+	node
+	d *pipeline.DeleteNode
+
+	fieldsDeleted *expvar.Int
+	tagsDeleted   *expvar.Int
+}
+
+// Create a new  DeleteNode which applies a transformation func to each point in a stream and returns a single point.
+func newDeleteNode(et *ExecutingTask, n *pipeline.DeleteNode, l *log.Logger) (*DeleteNode, error) {
+	dn := &DeleteNode{
+		node: node{Node: n, et: et, logger: l},
+		d:    n,
+	}
+	dn.node.runF = dn.runDelete
+	return dn, nil
+}
+
+func (e *DeleteNode) runDelete(snapshot []byte) error {
+	e.fieldsDeleted = &expvar.Int{}
+	e.tagsDeleted = &expvar.Int{}
+
+	e.statMap.Set(statsFieldsDeleted, e.fieldsDeleted)
+	e.statMap.Set(statsTagsDeleted, e.tagsDeleted)
+	switch e.Provides() {
+	case pipeline.StreamEdge:
+		for p, ok := e.ins[0].NextPoint(); ok; p, ok = e.ins[0].NextPoint() {
+			e.timer.Start()
+			p.Fields, p.Tags = e.doDeletes(p.Fields, p.Tags)
+			e.timer.Stop()
+			for _, child := range e.outs {
+				err := child.CollectPoint(p)
+				if err != nil {
+					return err
+				}
+			}
+		}
+	case pipeline.BatchEdge:
+		for b, ok := e.ins[0].NextBatch(); ok; b, ok = e.ins[0].NextBatch() {
+			e.timer.Start()
+			for i := range b.Points {
+				b.Points[i].Fields, b.Points[i].Tags = e.doDeletes(b.Points[i].Fields, b.Points[i].Tags)
+			}
+			e.timer.Stop()
+			for _, child := range e.outs {
+				err := child.CollectBatch(b)
+				if err != nil {
+					return err
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func (d *DeleteNode) doDeletes(fields models.Fields, tags models.Tags) (models.Fields, models.Tags) {
+	newFields := fields
+	fieldsCopied := false
+	for _, field := range d.d.Fields {
+		if _, ok := fields[field]; ok {
+			if !fieldsCopied {
+				newFields = newFields.Copy()
+				fieldsCopied = true
+			}
+			d.fieldsDeleted.Add(1)
+			delete(newFields, field)
+		}
+	}
+	newTags := tags
+	tagsCopied := false
+	for _, tag := range d.d.Tags {
+		if _, ok := tags[tag]; ok {
+			if !tagsCopied {
+				newTags = newTags.Copy()
+				tagsCopied = true
+			}
+			d.tagsDeleted.Add(1)
+			delete(newTags, tag)
+		}
+	}
+	return newFields, newTags
+}

edge.go

@@ -88,11 +88,11 @@ type edgeStat struct {
 	collected int64
 	emitted   int64
 	tags      models.Tags
-	dims      []string
+	dims      models.Dimensions
 }
 
 // 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)) {
+func (e *Edge) readGroupStats(f func(group models.GroupID, collected, emitted int64, tags models.Tags, dims models.Dimensions)) {
 	e.groupMu.RLock()
 	defer e.groupMu.RUnlock()
 	for group, stats := range e.groupStats {
@@ -188,7 +188,7 @@ func (e *Edge) CollectBatch(b models.Batch) error {
 }
 
 // Increment the emitted count of the group for this edge.
-func (e *Edge) incEmitted(group models.GroupID, tags models.Tags, dims []string, count int64) {
+func (e *Edge) incEmitted(group models.GroupID, tags models.Tags, dims models.Dimensions, count int64) {
 	// we are "manually" calling Unlock() and not using defer, because this method is called
 	// in hot locations (NextPoint/CollectPoint) and defer have some performance penalty
 	e.groupMu.Lock()
@@ -208,7 +208,7 @@ func (e *Edge) incEmitted(group models.GroupID, tags models.Tags, dims []string,
 }
 
 // Increment the collected count of the group for this edge.
-func (e *Edge) incCollected(group models.GroupID, tags models.Tags, dims []string, count int64) {
+func (e *Edge) incCollected(group models.GroupID, tags models.Tags, dims models.Dimensions, count int64) {
 	// we are "manually" calling Unlock() and not using defer, because this method is called
 	// in hot locations (NextPoint/CollectPoint) and defer have some performance penalty
 	e.groupMu.Lock()

edge_test.go

@@ -11,18 +11,19 @@ import (
 )
 
 func Benchmark_CollectPoint(b *testing.B) {
+	name := "point"
 	b.ReportAllocs()
 	ls := &logService{}
 	e := newEdge("BCollectPoint", "parent", "child", pipeline.StreamEdge, defaultEdgeBufferSize, ls)
 	p := models.Point{
-		Name: "point",
+		Name: name,
 		Tags: models.Tags{
 			"tag1": "value1",
 			"tag2": "value2",
 			"tag3": "value3",
 			"tag4": "value4",
 		},
-		Group: models.NilGroup,
+		Group: models.ToGroupID(name, nil, models.Dimensions{}),
 		Fields: models.Fields{
 			"field1": 42,
 			"field2": 4.2,

eval.go

@@ -24,6 +24,7 @@ type EvalNode struct {
 	expressionsByGroup map[models.GroupID][]stateful.Expression
 	refVarList         [][]string
 	scopePool          stateful.ScopePool
+	tags               map[string]bool
 
 	evalErrors *expvar.Int
 }
@@ -55,6 +56,14 @@ func newEvalNode(et *ExecutingTask, n *pipeline.EvalNode, l *log.Logger) (*EvalN
 	// Create a single pool for the combination of all expressions
 	en.scopePool = stateful.NewScopePool(stateful.FindReferenceVariables(expressions...))
 
+	// Create map of tags
+	if l := len(n.TagsList); l > 0 {
+		en.tags = make(map[string]bool, l)
+		for _, tag := range n.TagsList {
+			en.tags[tag] = true
+		}
+	}
+
 	en.node.runF = en.runEval
 	return en, nil
 }
@@ -67,7 +76,7 @@ func (e *EvalNode) runEval(snapshot []byte) error {
 		var err error
 		for p, ok := e.ins[0].NextPoint(); ok; p, ok = e.ins[0].NextPoint() {
 			e.timer.Start()
-			p.Fields, err = e.eval(p.Time, p.Group, p.Fields, p.Tags)
+			p.Fields, p.Tags, err = e.eval(p.Time, p.Group, p.Fields, p.Tags)
 			if err != nil {
 				e.evalErrors.Add(1)
 				if !e.e.QuiteFlag {
@@ -91,7 +100,7 @@ func (e *EvalNode) runEval(snapshot []byte) error {
 			e.timer.Start()
 			for i := 0; i < len(b.Points); {
 				p := b.Points[i]
-				b.Points[i].Fields, err = e.eval(p.Time, b.Group, p.Fields, p.Tags)
+				b.Points[i].Fields, b.Points[i].Tags, err = e.eval(p.Time, b.Group, p.Fields, p.Tags)
 				if err != nil {
 					e.evalErrors.Add(1)
 					if !e.e.QuiteFlag {
@@ -115,7 +124,7 @@ func (e *EvalNode) runEval(snapshot []byte) error {
 	return nil
 }
 
-func (e *EvalNode) eval(now time.Time, group models.GroupID, fields models.Fields, tags map[string]string) (models.Fields, error) {
+func (e *EvalNode) eval(now time.Time, group models.GroupID, fields models.Fields, tags models.Tags) (models.Fields, models.Tags, error) {
 	vars := e.scopePool.Get()
 	defer e.scopePool.Put(vars)
 	expressions, ok := e.expressionsByGroup[group]
@@ -129,23 +138,38 @@ func (e *EvalNode) eval(now time.Time, group models.GroupID, fields models.Field
 	for i, expr := range expressions {
 		err := fillScope(vars, e.refVarList[i], now, fields, tags)
 		if err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 		v, err := expr.Eval(vars)
 		if err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 		name := e.e.AsList[i]
 		vars.Set(name, v)
 	}
+	newTags := tags
+	if len(e.tags) > 0 {
+		newTags = newTags.Copy()
+		for tag := range e.tags {
+			v, err := vars.Get(tag)
+			if err != nil {
+				return nil, nil, err
+			}
+			if s, ok := v.(string); !ok {
+				return nil, nil, fmt.Errorf("result of a tag expression must be of type string, got %T", v)
+			} else {
+				newTags[tag] = s
+			}
+		}
+	}
 	var newFields models.Fields
 	if e.e.KeepFlag {
 		if l := len(e.e.KeepList); l != 0 {
 			newFields = make(models.Fields, l)
 			for _, f := range e.e.KeepList {
 				v, err := vars.Get(f)
 				if err != nil {
-					return nil, err
+					return nil, nil, err
 				}
 				newFields[f] = v
 			}
@@ -157,20 +181,23 @@ func (e *EvalNode) eval(now time.Time, group models.GroupID, fields models.Field
 			for _, f := range e.e.AsList {
 				v, err := vars.Get(f)
 				if err != nil {
-					return nil, err
+					return nil, nil, err
 				}
 				newFields[f] = v
 			}
 		}
 	} else {
-		newFields = make(models.Fields, len(e.e.AsList))
+		newFields = make(models.Fields, len(e.e.AsList)-len(e.tags))
 		for _, f := range e.e.AsList {
+			if e.tags[f] {
+				continue
+			}
 			v, err := vars.Get(f)
 			if err != nil {
-				return nil, err
+				return nil, nil, err
 			}
 			newFields[f] = v
 		}
 	}
-	return newFields, nil
+	return newFields, newTags, nil
 }