forked from influxdata/kapacitor
-
Notifications
You must be signed in to change notification settings - Fork 0
/
sample.go
52 lines (46 loc) · 993 Bytes
/
sample.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
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 FromNode.Truncate, QueryNode.GroupBy time or WindowNode.Align
// for ensuring data is aligned with a boundary.
type SampleNode struct {
chainnode
// Keep every N point or batch
// tick:ignore
N int64
// Keep one point or batch every Duration
// tick:ignore
Duration time.Duration
}
func newSampleNode(wants EdgeType, rate interface{}) *SampleNode {
var n int64
var d time.Duration
switch r := rate.(type) {
case int64:
n = r
case time.Duration:
d = r
default:
panic("must pass int64 or duration to new sample node")
}
return &SampleNode{
chainnode: newBasicChainNode("sample", wants, wants),
N: n,
Duration: d,
}
}