@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
go 239 lines 7.67 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 package beanstalk
4
5 import (
6 "errors"
7 "fmt"
8 "strconv"
9 "strings"
10
11 "github.com/netdata/netdata/go/plugins/logger"
12 "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/socket"
13
14 "gopkg.in/yaml.v2"
15 )
16
17 type beanstalkConn interface {
18 connect() error
19 disconnect() error
20 queryStats() (*beanstalkdStats, error)
21 queryListTubes() ([]string, error)
22 queryStatsTube(string) (*tubeStats, error)
23 }
24
25 // https://github.com/beanstalkd/beanstalkd/blob/91c54fc05dc759ef27459ce4383934e1a4f2fb4b/doc/protocol.txt#L553
26 type beanstalkdStats struct {
27 CurrentJobsUrgent int64 `yaml:"current-jobs-urgent" stm:"current-jobs-urgent"`
28 CurrentJobsReady int64 `yaml:"current-jobs-ready" stm:"current-jobs-ready"`
29 CurrentJobsReserved int64 `yaml:"current-jobs-reserved" stm:"current-jobs-reserved"`
30 CurrentJobsDelayed int64 `yaml:"current-jobs-delayed" stm:"current-jobs-delayed"`
31 CurrentJobsBuried int64 `yaml:"current-jobs-buried" stm:"current-jobs-buried"`
32 CmdPut int64 `yaml:"cmd-put" stm:"cmd-put"`
33 CmdPeek int64 `yaml:"cmd-peek" stm:"cmd-peek"`
34 CmdPeekReady int64 `yaml:"cmd-peek-ready" stm:"cmd-peek-ready"`
35 CmdPeekDelayed int64 `yaml:"cmd-peek-delayed" stm:"cmd-peek-delayed"`
36 CmdPeekBuried int64 `yaml:"cmd-peek-buried" stm:"cmd-peek-buried"`
37 CmdReserve int64 `yaml:"cmd-reserve" stm:"cmd-reserve"`
38 CmdReserveWithTimeout int64 `yaml:"cmd-reserve-with-timeout" stm:"cmd-reserve-with-timeout"`
39 CmdTouch int64 `yaml:"cmd-touch" stm:"cmd-touch"`
40 CmdUse int64 `yaml:"cmd-use" stm:"cmd-use"`
41 CmdWatch int64 `yaml:"cmd-watch" stm:"cmd-watch"`
42 CmdIgnore int64 `yaml:"cmd-ignore" stm:"cmd-ignore"`
43 CmdDelete int64 `yaml:"cmd-delete" stm:"cmd-delete"`
44 CmdRelease int64 `yaml:"cmd-release" stm:"cmd-release"`
45 CmdBury int64 `yaml:"cmd-bury" stm:"cmd-bury"`
46 CmdKick int64 `yaml:"cmd-kick" stm:"cmd-kick"`
47 CmdStats int64 `yaml:"cmd-stats" stm:"cmd-stats"`
48 CmdStatsJob int64 `yaml:"cmd-stats-job" stm:"cmd-stats-job"`
49 CmdStatsTube int64 `yaml:"cmd-stats-tube" stm:"cmd-stats-tube"`
50 CmdListTubes int64 `yaml:"cmd-list-tubes" stm:"cmd-list-tubes"`
51 CmdListTubeUsed int64 `yaml:"cmd-list-tube-used" stm:"cmd-list-tube-used"`
52 CmdListTubesWatched int64 `yaml:"cmd-list-tubes-watched" stm:"cmd-list-tubes-watched"`
53 CmdPauseTube int64 `yaml:"cmd-pause-tube" stm:"cmd-pause-tube"`
54 JobTimeouts int64 `yaml:"job-timeouts" stm:"job-timeouts"`
55 TotalJobs int64 `yaml:"total-jobs" stm:"total-jobs"`
56 CurrentTubes int64 `yaml:"current-tubes" stm:"current-tubes"`
57 CurrentConnections int64 `yaml:"current-connections" stm:"current-connections"`
58 CurrentProducers int64 `yaml:"current-producers" stm:"current-producers"`
59 CurrentWorkers int64 `yaml:"current-workers" stm:"current-workers"`
60 CurrentWaiting int64 `yaml:"current-waiting" stm:"current-waiting"`
61 TotalConnections int64 `yaml:"total-connections" stm:"total-connections"`
62 RusageUtime float64 `yaml:"rusage-utime" stm:"rusage-utime,1000,1"`
63 RusageStime float64 `yaml:"rusage-stime" stm:"rusage-stime,1000,1"`
64 Uptime int64 `yaml:"uptime" stm:"uptime"`
65 BinlogRecordsWritten int64 `yaml:"binlog-records-written" stm:"binlog-records-written"`
66 BinlogRecordsMigrated int64 `yaml:"binlog-records-migrated" stm:"binlog-records-migrated"`
67 }
68
69 // https://github.com/beanstalkd/beanstalkd/blob/91c54fc05dc759ef27459ce4383934e1a4f2fb4b/doc/protocol.txt#L497
70 type tubeStats struct {
71 Name string `yaml:"name"`
72 CurrentJobsUrgent int64 `yaml:"current-jobs-urgent" stm:"current-jobs-urgent"`
73 CurrentJobsReady int64 `yaml:"current-jobs-ready" stm:"current-jobs-ready"`
74 CurrentJobsReserved int64 `yaml:"current-jobs-reserved" stm:"current-jobs-reserved"`
75 CurrentJobsDelayed int64 `yaml:"current-jobs-delayed" stm:"current-jobs-delayed"`
76 CurrentJobsBuried int64 `yaml:"current-jobs-buried" stm:"current-jobs-buried"`
77 TotalJobs int64 `yaml:"total-jobs" stm:"total-jobs"`
78 CurrentUsing int64 `yaml:"current-using" stm:"current-using"`
79 CurrentWaiting int64 `yaml:"current-waiting" stm:"current-waiting"`
80 CurrentWatching int64 `yaml:"current-watching" stm:"current-watching"`
81 Pause float64 `yaml:"pause" stm:"pause"`
82 CmdDelete int64 `yaml:"cmd-delete" stm:"cmd-delete"`
83 CmdPauseTube int64 `yaml:"cmd-pause-tube" stm:"cmd-pause-tube"`
84 PauseTimeLeft float64 `yaml:"pause-time-left" stm:"pause-time-left"`
85 }
86
87 func newBeanstalkConn(conf Config, log *logger.Logger) beanstalkConn {
88 return &beanstalkClient{
89 Logger: log,
90 client: socket.New(socket.Config{
91 Address: conf.Address,
92 Timeout: conf.Timeout.Duration(),
93 MaxReadLines: 2000,
94 TLSConf: nil,
95 }),
96 }
97 }
98
99 const (
100 cmdQuit = "quit"
101 cmdStats = "stats"
102 cmdListTubes = "list-tubes"
103 cmdStatsTube = "stats-tube"
104 )
105
106 type beanstalkClient struct {
107 *logger.Logger
108
109 client socket.Client
110 }
111
112 func (c *beanstalkClient) connect() error {
113 return c.client.Connect()
114 }
115
116 func (c *beanstalkClient) disconnect() error {
117 _, _, _ = c.query(cmdQuit)
118 return c.client.Disconnect()
119 }
120
121 func (c *beanstalkClient) queryStats() (*beanstalkdStats, error) {
122 cmd := cmdStats
123
124 resp, data, err := c.query(cmd)
125 if err != nil {
126 return nil, err
127 }
128 if resp != "OK" {
129 return nil, fmt.Errorf("command '%s' bad response: %s", cmd, resp)
130 }
131
132 var stats beanstalkdStats
133
134 if err := yaml.Unmarshal(data, &stats); err != nil {
135 return nil, err
136 }
137
138 return &stats, nil
139 }
140
141 func (c *beanstalkClient) queryListTubes() ([]string, error) {
142 cmd := cmdListTubes
143
144 resp, data, err := c.query(cmd)
145 if err != nil {
146 return nil, err
147 }
148 if resp != "OK" {
149 return nil, fmt.Errorf("command '%s' bad response: %s", cmd, resp)
150 }
151
152 var tubes []string
153
154 if err := yaml.Unmarshal(data, &tubes); err != nil {
155 return nil, err
156 }
157
158 return tubes, nil
159 }
160
161 func (c *beanstalkClient) queryStatsTube(tubeName string) (*tubeStats, error) {
162 cmd := fmt.Sprintf("%s %s", cmdStatsTube, tubeName)
163
164 resp, data, err := c.query(cmd)
165 if err != nil {
166 return nil, err
167 }
168 if resp == "NOT_FOUND" {
169 return nil, nil
170 }
171 if resp != "OK" {
172 return nil, fmt.Errorf("command '%s' bad response: %s", cmd, resp)
173 }
174
175 var stats tubeStats
176 if err := yaml.Unmarshal(data, &stats); err != nil {
177 return nil, err
178 }
179
180 return &stats, nil
181 }
182
183 func (c *beanstalkClient) query(command string) (string, []byte, error) {
184 c.Debugf("executing command: %s", command)
185
186 var (
187 resp string
188 body []byte
189 length int
190 err error
191 )
192
193 if err := c.client.Command(command+"\r\n", func(line []byte) (bool, error) {
194 if resp == "" {
195 s := string(line)
196 c.Debugf("command '%s' response: '%s'", command, s)
197
198 resp, length, err = parseResponseLine(s)
199 if err != nil {
200 return false, fmt.Errorf("command '%s' line '%s': %v", command, s, err)
201 }
202
203 return resp == "OK", nil
204 }
205
206 body = append(body, line...)
207 body = append(body, '\n')
208
209 return len(body) < length, nil
210 }); err != nil {
211 return "", nil, fmt.Errorf("command '%s': %v", command, err)
212 }
213
214 return resp, body, nil
215 }
216
217 func parseResponseLine(line string) (string, int, error) {
218 parts := strings.Fields(line)
219 if len(parts) == 0 {
220 return "", 0, errors.New("empty response")
221 }
222
223 resp := parts[0]
224
225 if resp != "OK" {
226 return resp, 0, nil
227 }
228
229 if len(parts) < 2 {
230 return "", 0, errors.New("missing bytes count")
231 }
232
233 length, err := strconv.Atoi(parts[1])
234 if err != nil {
235 return "", 0, errors.New("invalid bytes count")
236 }
237
238 return resp, length, nil
239 }