forked from influxdata/kapacitor
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request influxdata#45 from influxdb/nc-issue#44
Add SampleNode and fix Grouping in Where node
- Loading branch information
Showing
12 changed files
with
254 additions
and
41 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,52 @@ | ||
| package pipeline | ||
|
|
||
| import ( | ||
| "time" | ||
| ) | ||
|
|
||
| // Sample points or batches. | ||
| // One point will be emitted every count or duration specified. | ||
| // | ||
| // Example: | ||
| // stream. | ||
| // .sample(3) | ||
| // | ||
| // Keep every third data point or batch. | ||
| // | ||
| // Example: | ||
| // stream. | ||
| // .sample(10s) | ||
| // | ||
| // Keep only samples that land on the 10s boundary. | ||
| // See WindowNode.Align or BatchNode.GroupBy time, | ||
| // for ensuring data is aligned with a boundary. | ||
| type SampleNode struct { | ||
| chainnode | ||
|
|
||
| // Keep every Count point or batch | ||
| // tick:ignore | ||
| Count int64 | ||
|
|
||
| // Keep one point or batch every Duration | ||
| // tick:ignore | ||
| Duration time.Duration | ||
| } | ||
|
|
||
| func newSampleNode(wants EdgeType, rate interface{}) *SampleNode { | ||
| var c int64 | ||
| var d time.Duration | ||
| switch r := rate.(type) { | ||
| case int64: | ||
| c = r | ||
| case time.Duration: | ||
| d = r | ||
| default: | ||
| panic("must pass int64 or duration to new sample node") | ||
| } | ||
|
|
||
| return &SampleNode{ | ||
| chainnode: newBasicChainNode("sample", wants, wants), | ||
| Count: c, | ||
| Duration: d, | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,73 @@ | ||
| package kapacitor | ||
|
|
||
| import ( | ||
| "errors" | ||
| "time" | ||
|
|
||
| "github.com/influxdb/kapacitor/models" | ||
| "github.com/influxdb/kapacitor/pipeline" | ||
| ) | ||
|
|
||
| type SampleNode struct { | ||
| node | ||
| s *pipeline.SampleNode | ||
|
|
||
| counts map[models.GroupID]int64 | ||
| duration time.Duration | ||
| } | ||
|
|
||
| // Create a new SampleNode which filters data from a source. | ||
| func newSampleNode(et *ExecutingTask, n *pipeline.SampleNode) (*SampleNode, error) { | ||
| sn := &SampleNode{ | ||
| node: node{Node: n, et: et}, | ||
| s: n, | ||
| counts: make(map[models.GroupID]int64), | ||
| duration: n.Duration, | ||
| } | ||
| sn.node.runF = sn.runSample | ||
| if n.Duration == 0 && n.Count == 0 { | ||
| return nil, errors.New("invalid sample rate: must be positive integer or duration") | ||
| } | ||
| return sn, nil | ||
| } | ||
|
|
||
| func (s *SampleNode) runSample() error { | ||
| switch s.Wants() { | ||
| case pipeline.StreamEdge: | ||
| for p, ok := s.ins[0].NextPoint(); ok; p, ok = s.ins[0].NextPoint() { | ||
| if s.shouldKeep(p.Group, p.Time) { | ||
| for _, child := range s.outs { | ||
| err := child.CollectPoint(p) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| } | ||
| } | ||
| } | ||
| case pipeline.BatchEdge: | ||
| for b, ok := s.ins[0].NextBatch(); ok; b, ok = s.ins[0].NextBatch() { | ||
| if s.shouldKeep(b.Group, b.TMax) { | ||
| for _, child := range s.outs { | ||
| err := child.CollectBatch(b) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (s *SampleNode) shouldKeep(group models.GroupID, t time.Time) bool { | ||
| if s.duration != 0 { | ||
| keepTime := t.Truncate(s.duration) | ||
| return t.Equal(keepTime) | ||
| } else { | ||
| count := s.counts[group] | ||
| keep := count%s.s.Count == 0 | ||
| count++ | ||
| s.counts[group] = count | ||
| return keep | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.