-
Notifications
You must be signed in to change notification settings - Fork 111
/
histogram.go
158 lines (146 loc) · 3.67 KB
/
histogram.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
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
package library
import (
"container/heap"
"errors"
"github.com/nytlabs/gojee" // jee
"github.com/nytlabs/streamtools/st/blocks" // blocks
"github.com/nytlabs/streamtools/st/util" // util
"time"
)
// specify those channels we're going to use to communicate with streamtools
type Histogram struct {
blocks.Block
queryrule chan chan interface{}
historule chan chan interface{}
inrule chan interface{}
inpoll chan interface{}
in chan interface{}
out chan interface{}
quit chan interface{}
}
// we need to build a simple factory so that streamtools can make new blocks of this kind
func NewHistogram() blocks.BlockInterface {
return &Histogram{}
}
func buildHistogram(histogram map[string]*PriorityQueue) interface{} {
var data interface{}
var buckets []interface{}
buckets = make([]interface{}, len(histogram))
i := 0
for k, pq := range histogram {
var bucket interface{}
bucket = map[string]interface{}{
"Count": float64(len(*pq)),
"Label": k,
}
buckets[i] = bucket
i++
}
data = map[string]interface{}{
"Histogram": buckets,
}
return data
}
// Setup is called once before running the block. We build up the channels and specify what kind of block this is.
func (b *Histogram) Setup() {
b.Kind = "Histogram"
b.Desc = "builds a non-stationary histogram of inbound messages for a specified path"
b.in = b.InRoute("in")
b.inrule = b.InRoute("rule")
b.queryrule = b.QueryRoute("rule")
b.historule = b.QueryRoute("histogram")
b.inpoll = b.InRoute("poll")
b.quit = b.Quit()
b.out = b.Broadcast()
}
// Run is the block's main loop. Here we listen on the different channels we set up.
func (b *Histogram) Run() {
var tree *jee.TokenTree
var path string
waitTimer := time.NewTimer(100 * time.Millisecond)
window := time.Duration(0)
histogram := map[string]*PriorityQueue{}
emptyByte := make([]byte, 0)
for {
select {
case ruleI := <-b.inrule:
// window
windowString, err := util.ParseString(ruleI, "Window")
if err != nil {
b.Error(err)
}
window, err = time.ParseDuration(windowString)
if err != nil {
b.Error(err)
}
path, err = util.ParseString(ruleI, "Path")
tree, err = util.BuildTokenTree(path)
if err != nil {
b.Error(err)
break
}
case <-b.quit:
// quit the block
return
case msg := <-b.in:
if tree == nil {
continue
}
v, err := jee.Eval(tree, msg)
if err != nil {
b.Error(err)
break
}
valueString, ok := v.(string)
if !ok {
b.Error(errors.New("nil value against" + path + " - ignoring"))
break
}
if pq, ok := histogram[valueString]; ok {
queueMessage := &PQMessage{
val: &emptyByte,
t: time.Now(),
}
heap.Push(pq, queueMessage)
} else {
pq := &PriorityQueue{}
heap.Init(pq)
histogram[valueString] = pq
queueMessage := &PQMessage{
val: &emptyByte,
t: time.Now(),
}
heap.Push(pq, queueMessage)
}
case <-waitTimer.C:
case <-b.inpoll:
// deal with a poll request
data := buildHistogram(histogram)
b.out <- data
case respChan := <-b.queryrule:
// deal with a query request
out := map[string]interface{}{
"Window": window.String(),
"Path": path,
}
respChan <- out
case respChan := <-b.historule:
data := buildHistogram(histogram)
respChan <- data
}
for _, pq := range histogram {
for {
pqMsg, diff := pq.PeekAndShift(time.Now(), window)
if pqMsg == nil {
// either the queue is empty, or it's not time to emit
if diff == 0 {
// then the queue is empty. Pause for 5 seconds before checking again
diff = time.Duration(500) * time.Millisecond
}
waitTimer.Reset(diff)
break
}
}
}
}
}