forked from elastic/beats
-
Notifications
You must be signed in to change notification settings - Fork 7
/
data.go
73 lines (64 loc) · 2.1 KB
/
data.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
package mntr
import (
"bufio"
"io"
"regexp"
"github.com/elastic/beats/libbeat/common"
s "github.com/elastic/beats/libbeat/common/schema"
c "github.com/elastic/beats/libbeat/common/schema/mapstrstr"
"github.com/elastic/beats/libbeat/logp"
)
var (
// Matches first the variable name, second the param itself
paramMatcher = regexp.MustCompile("([^\\s]+)\\s+(.*$)")
schema = s.Schema{
"version": c.Str("zk_version"),
"latency": s.Object{
"avg": c.Int("zk_avg_latency"),
"min": c.Int("zk_min_latency"),
"max": c.Int("zk_max_latency"),
},
"packets": s.Object{
"received": c.Int("zk_packets_received"),
"sent": c.Int("zk_packets_sent"),
},
"num_alive_connections": c.Int("zk_num_alive_connections"),
"outstanding_requests": c.Int("zk_outstanding_requests"),
"server_state": c.Str("zk_server_state"),
"znode_count": c.Int("zk_znode_count"),
"watch_count": c.Int("zk_watch_count"),
"ephemerals_count": c.Int("zk_ephemerals_count"),
"approximate_data_size": c.Int("zk_approximate_data_size"),
}
schemaLeader = s.Schema{
"followers": c.Int("zk_followers"),
"synced_followers": c.Int("zk_synced_followers"),
"pending_syncs": c.Int("zk_pending_syncs"),
}
schemaUnix = s.Schema{
"open_file_descriptor_count": c.Int("zk_open_file_descriptor_count"),
"max_file_descriptor_count": c.Int("zk_max_file_descriptor_count"),
}
)
func eventMapping(response io.Reader) common.MapStr {
fullEvent := map[string]interface{}{}
scanner := bufio.NewScanner(response)
// Iterate through all events to gather data
for scanner.Scan() {
if match := paramMatcher.FindStringSubmatch(scanner.Text()); len(match) == 3 {
fullEvent[match[1]] = match[2]
} else {
logp.Warn("Unexpected line in mntr output: %s", scanner.Text())
}
}
event, _ := schema.Apply(fullEvent)
// only exposed by the Leader
if _, ok := fullEvent["zk_followers"]; ok {
schemaLeader.ApplyTo(event, fullEvent)
}
// only available on Unix platforms
if _, ok := fullEvent["open_file_descriptor_count"]; ok {
schemaUnix.ApplyTo(event, fullEvent)
}
return event
}