@cryptotaxi247 / netdata-1 / commits / 6dc64510b

add go.d/gearman (#18294)

Co-authored-by: ilyam8 <22274335+ilyam8@users.noreply.github.com> Co-authored-by: Netdata bot <43409846+netdatabot@users.noreply.github.com>

Ilya Mashchenko committed Aug 9, 2024 at 18:50 UTC 6dc64510b3ed6c3186d22844ad2c9d161c4d188c
19 files changed +1143 -14
src/collectors/python.d.plugin/python.d.conf
+1 -1
@@ -34,7 +34,6 @@ gc_interval: 300
34 # this is just an example
35 example: no
36 # exim: yes
37 -# gearman: yes
37 go_expvar: no
38 # haproxy: yes
39 # monit: yes
@@ -63,6 +62,7 @@ beanstalk: no # Removed (replaced with go.d/beanstalk).
62 elasticsearch: no # Removed (replaced with go.d/elasticsearch).
63 fail2ban: no # Removed (replaced with go.d/fail2ban).
64 freeradius: no # Removed (replaced with go.d/freeradius).
65 +gearman: no # Removed (replaced with go.d/gearman).
66 hddtemp: no # Removed (replaced with go.d/hddtemp).
67 hpssa: no # Removed (replaced with go.d/hpssa).
68 icecast: no # Removed (replaced with go.d/icecast)
src/go/plugin/go.d/README.md
+1
@@ -78,6 +78,7 @@ see the appropriate collector readme.
78 | [filecheck](https://github.com/netdata/netdata/tree/master/src/go/plugin/go.d/modules/filecheck) | Files and Directories |
79 | [fluentd](https://github.com/netdata/netdata/tree/master/src/go/plugin/go.d/modules/fluentd) | Fluentd |
80 | [freeradius](https://github.com/netdata/netdata/tree/master/src/go/plugin/go.d/modules/freeradius) | FreeRADIUS |
81 +| [gearman](https://github.com/netdata/netdata/tree/master/src/go/plugin/go.d/modules/gearman) | Gearman |
82 | [haproxy](https://github.com/netdata/netdata/tree/master/src/go/plugin/go.d/modules/haproxy) | HAProxy |
83 | [hddtemp](https://github.com/netdata/netdata/tree/master/src/go/plugin/go.d/modules/hddtemp) | Disks temperature |
84 | [hdfs](https://github.com/netdata/netdata/tree/master/src/go/plugin/go.d/modules/hdfs) | HDFS |
src/go/plugin/go.d/config/go.d.conf
+1
@@ -43,6 +43,7 @@ modules:
43 # filecheck: yes
44 # fluentd: yes
45 # freeradius: yes
46 +# gearman: yes
47 # haproxy: yes
48 # hddtemp: yes
49 # hdfs: yes
src/go/plugin/go.d/config/go.d/gearman.conf new
+6
@@ -0,0 +1,6 @@
1 +## All available configuration options, their descriptions and default values:
2 +## https://github.com/netdata/netdata/tree/master/src/go/plugin/go.d/modules/gearman#readme
3 +
4 +#jobs:
5 +# - name: local
6 +# address: 127.0.0.1:4730
src/go/plugin/go.d/config/go.d/sd/docker.conf
+7
@@ -40,6 +40,8 @@ classify:
40 expr: '{{ or (eq .PrivatePort "5984") (match "sp" .Image "couchdb couchdb:*") }}'
41 - tags: "elasticsearch"
42 expr: '{{ or (eq .PrivatePort "9200") (match "sp" .Image "elasticsearch elasticsearch:* */elasticsearch */elasticsearch:* */opensearch */opensearch:*") }}'
43 + - tags: "gearman"
44 + expr: '{{ and (eq .PrivatePort "4730") (match "sp" .Image "*/gearmand */gearmand:*") }}'
45 - tags: "ipfs"
46 expr: '{{ and (eq .PrivatePort "5001") (match "sp" .Image "ipfs/kubo ipfs/kubo:*") }}'
47 - tags: "lighttpd"
@@ -134,6 +136,11 @@ compose:
136 username: admin
137 password: admin
138 {{ end -}}
139 + - selector: "gearman"
140 + template: |
141 + module: gearman
142 + name: docker_{{.Name}}
143 + address: {{.Address}}
144 - selector: "ipfs"
145 template: |
146 module: ipfs
src/go/plugin/go.d/config/go.d/sd/net_listeners.conf
+7
@@ -50,6 +50,8 @@ classify:
50 expr: '{{ and (eq .Port "24220") (glob .Cmdline "*fluentd*") }}'
51 - tags: "freeradius"
52 expr: '{{ and (eq .Port "18121") (eq .Comm "freeradius") }}'
53 + - tags: "gearman"
54 + expr: '{{ or (eq .Port "4730") (eq .Comm "gearmand") }}'
55 - tags: "geth"
56 expr: '{{ and (eq .Port "6060") (eq .Comm "geth") }}'
57 - tags: "haproxy"
@@ -248,6 +250,11 @@ compose:
250 address: {{.IPAddress}}
251 port: {{.Port}}
252 secret: adminsecret
253 + - selector: "gearman"
254 + template: |
255 + module: gearman
256 + name: local
257 + address: {{.Address}}
258 - selector: "geth"
259 template: |
260 module: geth
src/go/plugin/go.d/modules/gearman/charts.go new
+158
@@ -0,0 +1,158 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package gearman
4 +
5 +import (
6 + "fmt"
7 + "strings"
8 +
9 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/module"
10 +)
11 +
12 +const (
13 + prioQueuedJobsByActivity = module.Priority + iota
14 + prioQueuedJobsByPriority
15 +
16 + prioFunctionQueuedJobsByActivity
17 + prioFunctionQueuedJobsByPriority
18 + prioFunctionAvailableWorkers
19 +)
20 +
21 +var summaryCharts = module.Charts{
22 + chartQueuedJobsActivity.Copy(),
23 + chartQueuedJobsPriority.Copy(),
24 +}
25 +
26 +var (
27 + chartQueuedJobsActivity = module.Chart{
28 + ID: "queued_jobs_by_activity",
29 + Title: "Jobs Activity",
30 + Units: "jobs",
31 + Fam: "jobs",
32 + Ctx: "gearman.queued_jobs_activity",
33 + Priority: prioQueuedJobsByActivity,
34 + Type: module.Stacked,
35 + Dims: module.Dims{
36 + {ID: "total_jobs_running", Name: "running"},
37 + {ID: "total_jobs_waiting", Name: "waiting"},
38 + },
39 + }
40 + chartQueuedJobsPriority = module.Chart{
41 + ID: "queued_jobs_by_priority",
42 + Title: "Jobs Priority",
43 + Units: "jobs",
44 + Fam: "jobs",
45 + Ctx: "gearman.queued_jobs_priority",
46 + Priority: prioQueuedJobsByPriority,
47 + Type: module.Stacked,
48 + Dims: module.Dims{
49 + {ID: "total_high_priority_jobs", Name: "high"},
50 + {ID: "total_normal_priority_jobs", Name: "normal"},
51 + {ID: "total_low_priority_jobs", Name: "low"},
52 + },
53 + }
54 +)
55 +
56 +var functionStatusChartsTmpl = module.Charts{
57 + functionQueuedJobsActivityChartTmpl.Copy(),
58 + functionWorkersChartTmpl.Copy(),
59 +}
60 +
61 +var (
62 + functionQueuedJobsActivityChartTmpl = module.Chart{
63 + ID: "function_%s_queued_jobs_by_activity",
64 + Title: "Function Jobs Activity",
65 + Units: "jobs",
66 + Fam: "fn jobs",
67 + Ctx: "gearman.function_queued_jobs_activity",
68 + Priority: prioFunctionQueuedJobsByActivity,
69 + Type: module.Stacked,
70 + Dims: module.Dims{
71 + {ID: "function_%s_jobs_running", Name: "running"},
72 + {ID: "function_%s_jobs_waiting", Name: "waiting"},
73 + },
74 + }
75 + functionWorkersChartTmpl = module.Chart{
76 + ID: "function_%s_workers",
77 + Title: "Function Workers",
78 + Units: "workers",
79 + Fam: "fn workers",
80 + Ctx: "gearman.function_workers",
81 + Priority: prioFunctionAvailableWorkers,
82 + Type: module.Line,
83 + Dims: module.Dims{
84 + {ID: "function_%s_workers_available", Name: "available"},
85 + },
86 + }
87 +)
88 +
89 +var functionPriorityStatusChartsTmpl = module.Charts{
90 + functionQueuedJobsByPriorityChartTmpl.Copy(),
91 +}
92 +
93 +var (
94 + functionQueuedJobsByPriorityChartTmpl = module.Chart{
95 + ID: "prio_function_%s_queued_jobs_by_priority",
96 + Title: "Function Jobs Priority",
97 + Units: "jobs",
98 + Fam: "fn jobs",
99 + Ctx: "gearman.function_queued_jobs_priority",
100 + Priority: prioFunctionQueuedJobsByPriority,
101 + Type: module.Stacked,
102 + Dims: module.Dims{
103 + {ID: "function_%s_high_priority_jobs", Name: "high"},
104 + {ID: "function_%s_normal_priority_jobs", Name: "normal"},
105 + {ID: "function_%s_low_priority_jobs", Name: "low"},
106 + },
107 + }
108 +)
109 +
110 +func (g *Gearman) addFunctionStatusCharts(name string) {
111 + g.addFunctionCharts(name, functionStatusChartsTmpl.Copy())
112 +}
113 +
114 +func (g *Gearman) removeFunctionStatusCharts(name string) {
115 + px := fmt.Sprintf("function_%s_", cleanFunctionName(name))
116 + g.removeCharts(px)
117 +}
118 +
119 +func (g *Gearman) addFunctionPriorityStatusCharts(name string) {
120 + g.addFunctionCharts(name, functionPriorityStatusChartsTmpl.Copy())
121 +}
122 +
123 +func (g *Gearman) removeFunctionPriorityStatusCharts(name string) {
124 + px := fmt.Sprintf("prio_function_%s_", cleanFunctionName(name))
125 + g.removeCharts(px)
126 +}
127 +
128 +func (g *Gearman) addFunctionCharts(name string, charts *module.Charts) {
129 + charts = charts.Copy()
130 +
131 + for _, chart := range *charts {
132 + chart.ID = fmt.Sprintf(chart.ID, cleanFunctionName(name))
133 + chart.Labels = []module.Label{
134 + {Key: "function_name", Value: name},
135 + }
136 + for _, dim := range chart.Dims {
137 + dim.ID = fmt.Sprintf(dim.ID, name)
138 + }
139 + }
140 +
141 + if err := g.Charts().Add(*charts...); err != nil {
142 + g.Warning(err)
143 + }
144 +}
145 +
146 +func (g *Gearman) removeCharts(px string) {
147 + for _, chart := range *g.Charts() {
148 + if strings.HasPrefix(chart.ID, px) {
149 + chart.MarkRemove()
150 + chart.MarkNotCreated()
151 + }
152 + }
153 +}
154 +
155 +func cleanFunctionName(name string) string {
156 + r := strings.NewReplacer(".", "_", ",", "_", " ", "_")
157 + return r.Replace(name)
158 +}
src/go/plugin/go.d/modules/gearman/client.go new
+80
@@ -0,0 +1,80 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package gearman
4 +
5 +import (
6 + "bytes"
7 + "fmt"
8 + "strings"
9 +
10 + "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/socket"
11 +)
12 +
13 +type gearmanConn interface {
14 + connect() error
15 + disconnect()
16 + queryStatus() ([]byte, error)
17 + queryPriorityStatus() ([]byte, error)
18 +}
19 +
20 +func newGearmanConn(conf Config) gearmanConn {
21 + return &gearmanClient{conn: socket.New(socket.Config{
22 + Address: conf.Address,
23 + ConnectTimeout: conf.Timeout.Duration(),
24 + ReadTimeout: conf.Timeout.Duration(),
25 + WriteTimeout: conf.Timeout.Duration(),
26 + })}
27 +}
28 +
29 +type gearmanClient struct {
30 + conn socket.Client
31 +}
32 +
33 +func (c *gearmanClient) connect() error {
34 + return c.conn.Connect()
35 +}
36 +
37 +func (c *gearmanClient) disconnect() {
38 + _ = c.conn.Disconnect()
39 +}
40 +
41 +func (c *gearmanClient) queryStatus() ([]byte, error) {
42 + return c.query("status")
43 +}
44 +
45 +func (c *gearmanClient) queryPriorityStatus() ([]byte, error) {
46 + return c.query("prioritystatus")
47 +}
48 +
49 +func (c *gearmanClient) query(cmd string) ([]byte, error) {
50 + const limitReadLines = 10000
51 + var num int
52 + var err error
53 + var b bytes.Buffer
54 +
55 + clientErr := c.conn.Command(cmd+"\n", func(bs []byte) bool {
56 + s := string(bs)
57 +
58 + if strings.HasPrefix(s, "ERR") {
59 + err = fmt.Errorf("command '%s': %s", cmd, s)
60 + return false
61 + }
62 +
63 + b.WriteString(s)
64 + b.WriteByte('\n')
65 +
66 + if num++; num >= limitReadLines {
67 + err = fmt.Errorf("command '%s': read line limit exceeded (%d)", cmd, limitReadLines)
68 + return false
69 + }
70 + return !strings.HasPrefix(s, ".")
71 + })
72 + if clientErr != nil {
73 + return nil, fmt.Errorf("command '%s' client error: %v", cmd, clientErr)
74 + }
75 + if err != nil {
76 + return nil, err
77 + }
78 +
79 + return b.Bytes(), nil
80 +}
src/go/plugin/go.d/modules/gearman/collect.go new
+221
@@ -0,0 +1,221 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package gearman
4 +
5 +import (
6 + "bufio"
7 + "bytes"
8 + "errors"
9 + "fmt"
10 + "strconv"
11 + "strings"
12 +)
13 +
14 +func (g *Gearman) collect() (map[string]int64, error) {
15 + if g.conn == nil {
16 + conn, err := g.establishConn()
17 + if err != nil {
18 + return nil, err
19 + }
20 + g.conn = conn
21 + }
22 +
23 + status, err := g.conn.queryStatus()
24 + if err != nil {
25 + g.Cleanup()
26 + return nil, fmt.Errorf("couldn't query status: %v", err)
27 + }
28 +
29 + prioStatus, err := g.conn.queryPriorityStatus()
30 + if err != nil {
31 + g.Cleanup()
32 + return nil, fmt.Errorf("couldn't query priority status: %v", err)
33 + }
34 +
35 + mx := make(map[string]int64)
36 +
37 + if err := g.collectStatus(mx, status); err != nil {
38 + return nil, fmt.Errorf("couldn't collect status: %v", err)
39 + }
40 + if err := g.collectPriorityStatus(mx, prioStatus); err != nil {
41 + return nil, fmt.Errorf("couldn't collect priority status: %v", err)
42 + }
43 +
44 + return mx, nil
45 +
46 +}
47 +
48 +func (g *Gearman) collectStatus(mx map[string]int64, statusData []byte) error {
49 + /*
50 + Same output as the "gearadmin --status" command:
51 +
52 + FUNCTION\tTOTAL\tRUNNING\tAVAILABLE_WORKERS
53 +
54 + E.g.:
55 +
56 + prefix generic_worker4 78 78 500
57 + generic_worker2 78 78 500
58 + generic_worker3 0 0 760
59 + generic_worker1 0 0 500
60 + */
61 +
62 + seen := make(map[string]bool)
63 + var foundEnd bool
64 + sc := bufio.NewScanner(bytes.NewReader(statusData))
65 +
66 + mx["total_jobs_queued"] = 0
67 + mx["total_jobs_running"] = 0
68 + mx["total_jobs_waiting"] = 0
69 + mx["total_workers_avail"] = 0
70 +
71 + for sc.Scan() {
72 + line := strings.TrimSpace(sc.Text())
73 +
74 + if foundEnd = line == "."; foundEnd {
75 + break
76 + }
77 +
78 + parts := strings.Fields(line)
79 +
80 + // Gearman does not remove old tasks. We are only interested in tasks that have stats.
81 + if len(parts) < 4 {
82 + continue
83 + }
84 +
85 + name := strings.Join(parts[:len(parts)-3], "_")
86 + metrics := parts[len(parts)-3:]
87 +
88 + var queued, running, availWorkers int64
89 + var err error
90 +
91 + if queued, err = strconv.ParseInt(metrics[0], 10, 64); err != nil {
92 + return fmt.Errorf("couldn't parse queued count: %v", err)
93 + }
94 + if running, err = strconv.ParseInt(metrics[1], 10, 64); err != nil {
95 + return fmt.Errorf("couldn't parse running count: %v", err)
96 + }
97 + if availWorkers, err = strconv.ParseInt(metrics[2], 10, 64); err != nil {
98 + return fmt.Errorf("couldn't parse available count: %v", err)
99 + }
100 +
101 + px := fmt.Sprintf("function_%s_", name)
102 +
103 + waiting := queued - running
104 +
105 + mx[px+"jobs_queued"] = queued
106 + mx[px+"jobs_running"] = running
107 + mx[px+"jobs_waiting"] = waiting
108 + mx[px+"workers_available"] = availWorkers
109 +
110 + mx["total_jobs_queued"] += queued
111 + mx["total_jobs_running"] += running
112 + mx["total_jobs_waiting"] += waiting
113 + mx["total_workers_available"] += availWorkers
114 +
115 + seen[name] = true
116 + }
117 +
118 + if !foundEnd {
119 + return errors.New("unexpected status response")
120 + }
121 +
122 + for name := range seen {
123 + if !g.seenTasks[name] {
124 + g.seenTasks[name] = true
125 + g.addFunctionStatusCharts(name)
126 + }
127 + }
128 + for name := range g.seenTasks {
129 + if !seen[name] {
130 + delete(g.seenTasks, name)
131 + g.removeFunctionStatusCharts(name)
132 + }
133 + }
134 +
135 + return nil
136 +}
137 +
138 +func (g *Gearman) collectPriorityStatus(mx map[string]int64, prioStatusData []byte) error {
139 + /*
140 + Same output as the "gearadmin --priority-status" command:
141 +
142 + FUNCTION\tHIGH\tNORMAL\tLOW\tAVAILABLE_WORKERS
143 + */
144 +
145 + seen := make(map[string]bool)
146 + var foundEnd bool
147 + sc := bufio.NewScanner(bytes.NewReader(prioStatusData))
148 +
149 + mx["total_high_priority_jobs"] = 0
150 + mx["total_normal_priority_jobs"] = 0
151 + mx["total_low_priority_jobs"] = 0
152 +
153 + for sc.Scan() {
154 + line := strings.TrimSpace(sc.Text())
155 +
156 + if foundEnd = line == "."; foundEnd {
157 + break
158 + }
159 +
160 + parts := strings.Fields(line)
161 + if len(parts) < 5 {
162 + continue
163 + }
164 +
165 + name := strings.Join(parts[:len(parts)-4], "_")
166 + metrics := parts[len(parts)-4:]
167 +
168 + var high, normal, low int64
169 + var err error
170 +
171 + if high, err = strconv.ParseInt(metrics[0], 10, 64); err != nil {
172 + return fmt.Errorf("couldn't parse high count: %v", err)
173 + }
174 + if normal, err = strconv.ParseInt(metrics[1], 10, 64); err != nil {
175 + return fmt.Errorf("couldn't parse normal count: %v", err)
176 + }
177 + if low, err = strconv.ParseInt(metrics[2], 10, 64); err != nil {
178 + return fmt.Errorf("couldn't parse low count: %v", err)
179 + }
180 +
181 + px := fmt.Sprintf("function_%s_", name)
182 +
183 + mx[px+"high_priority_jobs"] = high
184 + mx[px+"normal_priority_jobs"] = normal
185 + mx[px+"low_priority_jobs"] = low
186 + mx["total_high_priority_jobs"] += high
187 + mx["total_normal_priority_jobs"] += normal
188 + mx["total_low_priority_jobs"] += low
189 +
190 + seen[name] = true
191 + }
192 +
193 + if !foundEnd {
194 + return errors.New("unexpected priority status response")
195 + }
196 +
197 + for name := range seen {
198 + if !g.seenPriorityTasks[name] {
199 + g.seenPriorityTasks[name] = true
200 + g.addFunctionPriorityStatusCharts(name)
201 + }
202 + }
203 + for name := range g.seenPriorityTasks {
204 + if !seen[name] {
205 + delete(g.seenPriorityTasks, name)
206 + g.removeFunctionPriorityStatusCharts(name)
207 + }
208 + }
209 +
210 + return nil
211 +}
212 +
213 +func (g *Gearman) establishConn() (gearmanConn, error) {
214 + conn := g.newConn(g.Config)
215 +
216 + if err := conn.connect(); err != nil {
217 + return nil, err
218 + }
219 +
220 + return conn, nil
221 +}
src/go/plugin/go.d/modules/gearman/config_schema.json new
+44
@@ -0,0 +1,44 @@
1 +{
2 + "jsonSchema": {
3 + "$schema": "http://json-schema.org/draft-07/schema#",
4 + "title": "Gearman collector configuration.",
5 + "type": "object",
6 + "properties": {
7 + "update_every": {
8 + "title": "Update every",
9 + "description": "Data collection interval, measured in seconds.",
10 + "type": "integer",
11 + "minimum": 1,
12 + "default": 1
13 + },
14 + "address": {
15 + "title": "Address",
16 + "description": "The IP address and port where the Gearman service listens for connections.",
17 + "type": "string",
18 + "default": "127.0.0.1:4730"
19 + },
20 + "timeout": {
21 + "title": "Timeout",
22 + "description": "Timeout for establishing a connection and communication (reading and writing) in seconds.",
23 + "type": "number",
24 + "minimum": 0.5,
25 + "default": 1
26 + }
27 + },
28 + "required": [
29 + "address"
30 + ],
31 + "additionalProperties": false,
32 + "patternProperties": {
33 + "^name$": {}
34 + }
35 + },
36 + "uiSchema": {
37 + "uiOptions": {
38 + "fullPage": true
39 + },
40 + "timeout": {
41 + "ui:help": "Accepts decimals for precise control (e.g., type 1.5 for 1.5 seconds)."
42 + }
43 + }
44 +}
src/go/plugin/go.d/modules/gearman/gearman.go new
+106
@@ -0,0 +1,106 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package gearman
4 +
5 +import (
6 + _ "embed"
7 + "errors"
8 + "time"
9 +
10 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/module"
11 + "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/web"
12 +)
13 +
14 +//go:embed "config_schema.json"
15 +var configSchema string
16 +
17 +func init() {
18 + module.Register("gearman", module.Creator{
19 + JobConfigSchema: configSchema,
20 + Create: func() module.Module { return New() },
21 + Config: func() any { return &Config{} },
22 + })
23 +}
24 +
25 +func New() *Gearman {
26 + return &Gearman{
27 + Config: Config{
28 + Address: "127.0.0.1:4730",
29 + Timeout: web.Duration(time.Second * 1),
30 + },
31 + newConn: newGearmanConn,
32 + charts: summaryCharts.Copy(),
33 + seenTasks: make(map[string]bool),
34 + seenPriorityTasks: make(map[string]bool),
35 + }
36 +}
37 +
38 +type Config struct {
39 + UpdateEvery int `yaml:"update_every,omitempty" json:"update_every"`
40 + Address string `yaml:"address" json:"address"`
41 + Timeout web.Duration `yaml:"timeout" json:"timeout"`
42 +}
43 +
44 +type Gearman struct {
45 + module.Base
46 + Config `yaml:",inline" json:""`
47 +
48 + charts *module.Charts
49 +
50 + newConn func(Config) gearmanConn
51 + conn gearmanConn
52 +
53 + seenTasks map[string]bool
54 + seenPriorityTasks map[string]bool
55 +}
56 +
57 +func (g *Gearman) Configuration() any {
58 + return g.Config
59 +}
60 +
61 +func (g *Gearman) Init() error {
62 + if g.Address == "" {
63 + g.Error("config: 'address' not set")
64 + return errors.New("address not set")
65 + }
66 +
67 + return nil
68 +}
69 +
70 +func (g *Gearman) Check() error {
71 + mx, err := g.collect()
72 + if err != nil {
73 + g.Error(err)
74 + return err
75 + }
76 +
77 + if len(mx) == 0 {
78 + return errors.New("no metrics collected")
79 + }
80 +
81 + return nil
82 +}
83 +
84 +func (g *Gearman) Charts() *module.Charts {
85 + return g.charts
86 +}
87 +
88 +func (g *Gearman) Collect() map[string]int64 {
89 + mx, err := g.collect()
90 + if err != nil {
91 + g.Error(err)
92 + }
93 +
94 + if len(mx) == 0 {
95 + return nil
96 + }
97 +
98 + return mx
99 +}
100 +
101 +func (g *Gearman) Cleanup() {
102 + if g.conn != nil {
103 + g.conn.disconnect()
104 + g.conn = nil
105 + }
106 +}
src/go/plugin/go.d/modules/gearman/gearman_test.go new
+326
@@ -0,0 +1,326 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package gearman
4 +
5 +import (
6 + "errors"
7 + "os"
8 + "testing"
9 +
10 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/module"
11 +
12 + "github.com/stretchr/testify/assert"
13 + "github.com/stretchr/testify/require"
14 +)
15 +
16 +var (
17 + dataConfigJSON, _ = os.ReadFile("testdata/config.json")
18 + dataConfigYAML, _ = os.ReadFile("testdata/config.yaml")
19 +
20 + dataStatus, _ = os.ReadFile("testdata/status.txt")
21 + dataPriorityStatus, _ = os.ReadFile("testdata/priority-status.txt")
22 +)
23 +
24 +func Test_testDataIsValid(t *testing.T) {
25 + for name, data := range map[string][]byte{
26 + "dataConfigJSON": dataConfigJSON,
27 + "dataConfigYAML": dataConfigYAML,
28 +
29 + "dataStatus": dataStatus,
30 + "dataPriorityStatus": dataPriorityStatus,
31 + } {
32 + require.NotNil(t, data, name)
33 + }
34 +}
35 +
36 +func TestGearman_ConfigurationSerialize(t *testing.T) {
37 + module.TestConfigurationSerialize(t, &Gearman{}, dataConfigJSON, dataConfigYAML)
38 +}
39 +
40 +func TestGearman_Init(t *testing.T) {
41 + tests := map[string]struct {
42 + config Config
43 + wantFail bool
44 + }{
45 + "success with default config": {
46 + wantFail: false,
47 + config: New().Config,
48 + },
49 + "fails if address not set": {
50 + wantFail: true,
51 + config: func() Config {
52 + conf := New().Config
53 + conf.Address = ""
54 + return conf
55 + }(),
56 + },
57 + }
58 +
59 + for name, test := range tests {
60 + t.Run(name, func(t *testing.T) {
61 + gear := New()
62 + gear.Config = test.config
63 +
64 + if test.wantFail {
65 + assert.Error(t, gear.Init())
66 + } else {
67 + assert.NoError(t, gear.Init())
68 + }
69 + })
70 + }
71 +}
72 +
73 +func TestGearman_Cleanup(t *testing.T) {
74 + tests := map[string]struct {
75 + prepare func() *Gearman
76 + }{
77 + "not initialized": {
78 + prepare: func() *Gearman {
79 + return New()
80 + },
81 + },
82 + "after check": {
83 + prepare: func() *Gearman {
84 + gear := New()
85 + gear.newConn = func(config Config) gearmanConn { return prepareMockOk() }
86 + _ = gear.Check()
87 + return gear
88 + },
89 + },
90 + "after collect": {
91 + prepare: func() *Gearman {
92 + gear := New()
93 + gear.newConn = func(config Config) gearmanConn { return prepareMockOk() }
94 + _ = gear.Collect()
95 + return gear
96 + },
97 + },
98 + }
99 +
100 + for name, test := range tests {
101 + t.Run(name, func(t *testing.T) {
102 + gear := test.prepare()
103 +
104 + assert.NotPanics(t, gear.Cleanup)
105 + })
106 + }
107 +}
108 +
109 +func TestGearman_Charts(t *testing.T) {
110 + assert.NotNil(t, New().Charts())
111 +}
112 +
113 +func TestGearman_Check(t *testing.T) {
114 + tests := map[string]struct {
115 + prepareMock func() *mockGearmanConn
116 + wantFail bool
117 + }{
118 + "success case": {
119 + wantFail: false,
120 + prepareMock: prepareMockOk,
121 + },
122 + "err on connect": {
123 + wantFail: true,
124 + prepareMock: prepareMockErrOnConnect,
125 + },
126 + "unexpected response": {
127 + wantFail: true,
128 + prepareMock: prepareMockUnexpectedResponse,
129 + },
130 + "empty response": {
131 + wantFail: false,
132 + prepareMock: prepareMockEmptyResponse,
133 + },
134 + }
135 +
136 + for name, test := range tests {
137 + t.Run(name, func(t *testing.T) {
138 + gear := New()
139 + mock := test.prepareMock()
140 + gear.newConn = func(config Config) gearmanConn { return mock }
141 +
142 + if test.wantFail {
143 + assert.Error(t, gear.Check())
144 + } else {
145 + assert.NoError(t, gear.Check())
146 + }
147 + })
148 + }
149 +}
150 +
151 +func TestGearman_Collect(t *testing.T) {
152 + tests := map[string]struct {
153 + prepareMock func() *mockGearmanConn
154 + wantMetrics map[string]int64
155 + wantCharts int
156 + disconnectBeforeCleanup bool
157 + disconnectAfterCleanup bool
158 + }{
159 + "success case": {
160 + prepareMock: prepareMockOk,
161 + disconnectBeforeCleanup: false,
162 + disconnectAfterCleanup: true,
163 + wantCharts: len(summaryCharts) + len(functionStatusChartsTmpl)*4 + len(functionPriorityStatusChartsTmpl)*4,
164 + wantMetrics: map[string]int64{
165 + "function_generic_worker1_high_priority_jobs": 10,
166 + "function_generic_worker1_jobs_queued": 4,
167 + "function_generic_worker1_jobs_running": 3,
168 + "function_generic_worker1_jobs_waiting": 1,
169 + "function_generic_worker1_low_priority_jobs": 12,
170 + "function_generic_worker1_normal_priority_jobs": 11,
171 + "function_generic_worker1_workers_available": 500,
172 + "function_generic_worker2_high_priority_jobs": 4,
173 + "function_generic_worker2_jobs_queued": 78,
174 + "function_generic_worker2_jobs_running": 78,
175 + "function_generic_worker2_jobs_waiting": 0,
176 + "function_generic_worker2_low_priority_jobs": 6,
177 + "function_generic_worker2_normal_priority_jobs": 5,
178 + "function_generic_worker2_workers_available": 500,
179 + "function_generic_worker3_high_priority_jobs": 7,
180 + "function_generic_worker3_jobs_queued": 2,
181 + "function_generic_worker3_jobs_running": 1,
182 + "function_generic_worker3_jobs_waiting": 1,
183 + "function_generic_worker3_low_priority_jobs": 9,
184 + "function_generic_worker3_normal_priority_jobs": 8,
185 + "function_generic_worker3_workers_available": 760,
186 + "function_prefix_generic_worker4_high_priority_jobs": 1,
187 + "function_prefix_generic_worker4_jobs_queued": 78,
188 + "function_prefix_generic_worker4_jobs_running": 78,
189 + "function_prefix_generic_worker4_jobs_waiting": 0,
190 + "function_prefix_generic_worker4_low_priority_jobs": 3,
191 + "function_prefix_generic_worker4_normal_priority_jobs": 2,
192 + "function_prefix_generic_worker4_workers_available": 500,
193 + "total_high_priority_jobs": 22,
194 + "total_jobs_queued": 162,
195 + "total_jobs_running": 160,
196 + "total_jobs_waiting": 2,
197 + "total_low_priority_jobs": 30,
198 + "total_normal_priority_jobs": 26,
199 + "total_workers_avail": 0,
200 + "total_workers_available": 2260,
201 + },
202 + },
203 + "unexpected response": {
204 + prepareMock: prepareMockUnexpectedResponse,
205 + disconnectBeforeCleanup: false,
206 + disconnectAfterCleanup: true,
207 + },
208 + "empty response": {
209 + prepareMock: prepareMockEmptyResponse,
210 + disconnectBeforeCleanup: false,
211 + disconnectAfterCleanup: true,
212 + wantCharts: len(summaryCharts),
213 + wantMetrics: map[string]int64{
214 + "total_high_priority_jobs": 0,
215 + "total_jobs_queued": 0,
216 + "total_jobs_running": 0,
217 + "total_jobs_waiting": 0,
218 + "total_low_priority_jobs": 0,
219 + "total_normal_priority_jobs": 0,
220 + "total_workers_avail": 0,
221 + },
222 + },
223 + "err on connect": {
224 + prepareMock: prepareMockErrOnConnect,
225 + disconnectBeforeCleanup: false,
226 + disconnectAfterCleanup: false,
227 + },
228 + "err on query status": {
229 + prepareMock: prepareMockErrOnQueryStatus,
230 + disconnectBeforeCleanup: true,
231 + disconnectAfterCleanup: true,
232 + },
233 + }
234 +
235 + for name, test := range tests {
236 + t.Run(name, func(t *testing.T) {
237 + gear := New()
238 + mock := test.prepareMock()
239 + gear.newConn = func(config Config) gearmanConn { return mock }
240 +
241 + mx := gear.Collect()
242 +
243 + require.Equal(t, test.wantMetrics, mx, "want metrics")
244 +
245 + if len(test.wantMetrics) > 0 {
246 + module.TestMetricsHasAllChartsDims(t, gear.Charts(), mx)
247 + assert.Equal(t, test.wantCharts, len(*gear.Charts()), "want charts")
248 + }
249 +
250 + assert.Equal(t, test.disconnectBeforeCleanup, mock.disconnectCalled, "disconnect before cleanup")
251 + gear.Cleanup()
252 + assert.Equal(t, test.disconnectAfterCleanup, mock.disconnectCalled, "disconnect after cleanup")
253 + })
254 + }
255 +}
256 +
257 +func prepareMockOk() *mockGearmanConn {
258 + return &mockGearmanConn{
259 + responseStatus: dataStatus,
260 + responsePriorityStatus: dataPriorityStatus,
261 + }
262 +}
263 +
264 +func prepareMockErrOnConnect() *mockGearmanConn {
265 + return &mockGearmanConn{
266 + errOnConnect: true,
267 + }
268 +}
269 +
270 +func prepareMockErrOnQueryStatus() *mockGearmanConn {
271 + return &mockGearmanConn{
272 + errOnQueryStatus: true,
273 + }
274 +}
275 +
276 +func prepareMockUnexpectedResponse() *mockGearmanConn {
277 + resp := []byte("Lorem ipsum dolor sit amet, consectetur adipiscing elit.")
278 + return &mockGearmanConn{
279 + responseStatus: resp,
280 + responsePriorityStatus: resp,
281 + }
282 +}
283 +
284 +func prepareMockEmptyResponse() *mockGearmanConn {
285 + return &mockGearmanConn{
286 + responseStatus: []byte("."),
287 + responsePriorityStatus: []byte("."),
288 + }
289 +}
290 +
291 +type mockGearmanConn struct {
292 + errOnConnect bool
293 +
294 + responseStatus []byte
295 + errOnQueryStatus bool
296 +
297 + responsePriorityStatus []byte
298 + errOnQueryPriorityStatus bool
299 +
300 + disconnectCalled bool
301 +}
302 +
303 +func (m *mockGearmanConn) connect() error {
304 + if m.errOnConnect {
305 + return errors.New("mock.connect() error")
306 + }
307 + return nil
308 +}
309 +
310 +func (m *mockGearmanConn) disconnect() {
311 + m.disconnectCalled = true
312 +}
313 +
314 +func (m *mockGearmanConn) queryStatus() ([]byte, error) {
315 + if m.errOnQueryStatus {
316 + return nil, errors.New("mock.queryStatus() error")
317 + }
318 + return m.responseStatus, nil
319 +}
320 +
321 +func (m *mockGearmanConn) queryPriorityStatus() ([]byte, error) {
322 + if m.errOnQueryPriorityStatus {
323 + return nil, errors.New("mock.queryPriorityStatus() error")
324 + }
325 + return m.responsePriorityStatus, nil
326 +}
src/go/plugin/go.d/modules/gearman/metadata.yaml new
+152
@@ -0,0 +1,152 @@
1 +plugin_name: go.d.plugin
2 +modules:
3 + - meta:
4 + id: collector-go.d.plugin-gearman
5 + plugin_name: go.d.plugin
6 + module_name: gearman
7 + monitored_instance:
8 + name: Gearman
9 + link: https://gearman.org/
10 + categories:
11 + - data-collection.distributed-computing-systems
12 + icon_filename: "gearman.png"
13 + related_resources:
14 + integrations:
15 + list: []
16 + info_provided_to_referring_integrations:
17 + description: ""
18 + keywords:
19 + - gearman
20 + most_popular: false
21 + overview:
22 + data_collection:
23 + metrics_description: |
24 + Monitors jobs activity, priority and available workers. It collects summary and function-specific statistics.
25 + method_description: |
26 + This collector connects to a Gearman instance via TCP socket and executes the following commands:
27 +
28 + - status
29 + - priority-status
30 + supported_platforms:
31 + include: []
32 + exclude: []
33 + multi_instance: true
34 + additional_permissions:
35 + description: ""
36 + default_behavior:
37 + auto_detection:
38 + description: |
39 + By default, it detects Gearman instances running on localhost that are listening on port 4730.
40 + limits:
41 + description: ""
42 + performance_impact:
43 + description: ""
44 + setup:
45 + prerequisites:
46 + list: []
47 + configuration:
48 + file:
49 + name: go.d/memcached.conf
50 + options:
51 + description: |
52 + The following options can be defined globally: update_every, autodetection_retry.
53 + folding:
54 + title: Config options
55 + enabled: true
56 + list:
57 + - name: update_every
58 + description: Data collection frequency.
59 + default_value: 1
60 + required: false
61 + - name: autodetection_retry
62 + description: Recheck interval in seconds. Zero means no recheck will be scheduled.
63 + default_value: 0
64 + required: false
65 + - name: address
66 + description: The IP address and port where the Gearman service listens for connections.
67 + default_value: 127.0.0.1:11211
68 + required: true
69 + - name: timeout
70 + description: Connection, read, and write timeout duration in seconds. The timeout includes name resolution.
71 + default_value: 1
72 + required: false
73 + examples:
74 + folding:
75 + title: Config
76 + enabled: true
77 + list:
78 + - name: Basic
79 + description: A basic example configuration.
80 + config: |
81 + jobs:
82 + - name: local
83 + address: 127.0.0.1:4730
84 + - name: Multi-instance
85 + description: |
86 + > **Note**: When you define multiple jobs, their names must be unique.
87 +
88 + Collecting metrics from local and remote instances.
89 + config: |
90 + jobs:
91 + - name: local
92 + address: 127.0.0.1:4730
93 +
94 + - name: remote
95 + address: 203.0.113.0:4730
96 + troubleshooting:
97 + problems:
98 + list: []
99 + alerts: []
100 + metrics:
101 + folding:
102 + title: Metrics
103 + enabled: false
104 + description: ""
105 + availability: []
106 + scopes:
107 + - name: global
108 + description: "These metrics refer to the entire monitored application."
109 + labels: []
110 + metrics:
111 + - name: gearman.queued_jobs_activity
112 + description: Jobs Activity
113 + unit: "jobs"
114 + chart_type: stacked
115 + dimensions:
116 + - name: running
117 + - name: waiting
118 + - name: gearman.queued_jobs_priority
119 + description: Jobs Priority
120 + unit: "jobs"
121 + chart_type: stacked
122 + dimensions:
123 + - name: high
124 + - name: normal
125 + - name: low
126 + - name: global
127 + description: "These metrics refer to the Function (task)."
128 + labels:
129 + - name: function_name
130 + description: Function name.
131 + metrics:
132 + - name: gearman.function_queued_jobs_activity
133 + description: Function Jobs Activity
134 + unit: "jobs"
135 + chart_type: stacked
136 + dimensions:
137 + - name: running
138 + - name: waiting
139 + - name: gearman.function_queued_jobs_priority
140 + description: Function Jobs Priority
141 + unit: "jobs"
142 + chart_type: stacked
143 + dimensions:
144 + - name: high
145 + - name: normal
146 + - name: low
147 + - name: gearman.function_workers
148 + description: Function Workers
149 + unit: "workers"
150 + chart_type: line
151 + dimensions:
152 + - name: available
src/go/plugin/go.d/modules/gearman/testdata/config.json new
+5
@@ -0,0 +1,5 @@
1 +{
2 + "update_every": 123,
3 + "address": "ok",
4 + "timeout": 123.123
5 +}
src/go/plugin/go.d/modules/gearman/testdata/config.yaml new
+3
@@ -0,0 +1,3 @@
1 +update_every: 123
2 +address: "ok"
3 +timeout: 123.123
src/go/plugin/go.d/modules/gearman/testdata/priority-status.txt new
+5
@@ -0,0 +1,5 @@
1 +prefix generic_worker4 1 2 3 500
2 +generic_worker2 4 5 6 500
3 +generic_worker3 7 8 9 760
4 +generic_worker1 10 11 12 500
5 +.
src/go/plugin/go.d/modules/gearman/testdata/status.txt new
+5
@@ -0,0 +1,5 @@
1 +prefix generic_worker4 78 78 500
2 +generic_worker2 78 78 500
3 +generic_worker3 2 1 760
4 +generic_worker1 4 3 500
5 +.
src/go/plugin/go.d/modules/init.go
+1
@@ -32,6 +32,7 @@ import (
32 _ "github.com/netdata/netdata/go/plugins/plugin/go.d/modules/filecheck"
33 _ "github.com/netdata/netdata/go/plugins/plugin/go.d/modules/fluentd"
34 _ "github.com/netdata/netdata/go/plugins/plugin/go.d/modules/freeradius"
35 + _ "github.com/netdata/netdata/go/plugins/plugin/go.d/modules/gearman"
36 _ "github.com/netdata/netdata/go/plugins/plugin/go.d/modules/geth"
37 _ "github.com/netdata/netdata/go/plugins/plugin/go.d/modules/haproxy"
38 _ "github.com/netdata/netdata/go/plugins/plugin/go.d/modules/hddtemp"
src/health/health.d/gearman.conf
+14 -13
@@ -1,14 +1,15 @@
1 +# you can disable an alarm notification by setting the 'to' line to: silent
2
2 - template: gearman_workers_queued
3 - on: gearman.single_job
4 - class: Latency
5 - type: Computing
6 -component: Gearman
7 - lookup: average -10m unaligned match-names of Pending
8 - units: workers
9 - every: 10s
10 - warn: $this > 30000
11 - delay: down 5m multiplier 1.5 max 1h
12 - summary: Gearman queued jobs
13 - info: Average number of queued jobs over the last 10 minutes
14 - to: sysadmin
3 +# template: gearman_function_waiting_jobs
4 +# on: gearman.function_queued_jobs_activity
5 +# class: Latency
6 +# type: Computing
7 +#component: Gearman
8 +# lookup: average -10m unaligned of waiting
9 +# units: jobs
10 +# every: 10s
11 +# warn: $this > 30000
12 +# delay: down 5m multiplier 1.5 max 1h
13 +# summary: Waiting jobs for ${label:task_name} function
14 +# info: Average number of waiting jobs for ${label:function_name} function over the last 10 minutes
15 +# to: sysadmin