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
|
// SPDX-License-Identifier: GPL-3.0-or-later
package zookeeper
import (
"bytes"
"fmt"
"unsafe"
"github.com/netdata/netdata/go/go.d.plugin/pkg/socket"
)
const limitReadLines = 2000
type zookeeperFetcher struct {
socket.Client
}
func (c *zookeeperFetcher) fetch(command string) (rows []string, err error) {
if err = c.Connect(); err != nil {
return nil, err
}
defer func() { _ = c.Disconnect() }()
var num int
clientErr := c.Command(command, func(b []byte) bool {
if !isZKLine(b) || isMntrLineOK(b) {
rows = append(rows, string(b))
}
if num += 1; num >= limitReadLines {
err = fmt.Errorf("read line limit exceeded (%d)", limitReadLines)
return false
}
return true
})
if clientErr != nil {
return nil, clientErr
}
if err != nil {
return nil, err
}
return rows, nil
}
func isZKLine(line []byte) bool {
return bytes.HasPrefix(line, []byte("zk_"))
}
func isMntrLineOK(line []byte) bool {
idx := bytes.LastIndexByte(line, '\t')
return idx > 0 && collectedZKKeys[unsafeString(line)[:idx]]
}
func unsafeString(b []byte) string {
return *((*string)(unsafe.Pointer(&b)))
}
var collectedZKKeys = map[string]bool{
"zk_num_alive_connections": true,
"zk_outstanding_requests": true,
"zk_min_latency": true,
"zk_avg_latency": true,
"zk_max_latency": true,
"zk_packets_received": true,
"zk_packets_sent": true,
"zk_open_file_descriptor_count": true,
"zk_max_file_descriptor_count": true,
"zk_znode_count": true,
"zk_ephemerals_count": true,
"zk_watch_count": true,
"zk_approximate_data_size": true,
"zk_server_state": true,
}
|
