@cryptotaxi247 / netdata-1 / commits / a1789a464

improvement(go.d/nats): add gatewayz metrics (#19266)

Ilya Mashchenko committed Dec 23, 2024 at 15:03 UTC a1789a464b525a576898c712192a5eb939ec5c2a
8 files changed +726 -152
src/go/plugin/go.d/collector/nats/cache.go new
+108
@@ -0,0 +1,108 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package nats
4 +
5 +func newCache() *cache {
6 + return &cache{
7 + accounts: make(accCache),
8 + routes: make(routeCache),
9 + inGateways: make(gwCache),
10 + outGateways: make(gwCache),
11 + }
12 +}
13 +
14 +type cache struct {
15 + accounts accCache
16 + routes routeCache
17 + inGateways gwCache
18 + outGateways gwCache
19 +}
20 +
21 +func (c *cache) resetUpdated() {
22 + for _, gw := range c.inGateways {
23 + gw.updated = false
24 + for _, conn := range gw.conns {
25 + conn.updated = false
26 + }
27 + }
28 + for _, gw := range c.outGateways {
29 + gw.updated = false
30 + for _, conn := range gw.conns {
31 + conn.updated = false
32 + }
33 + }
34 +}
35 +
36 +type (
37 + accCache map[string]*accCacheEntry
38 + accCacheEntry struct {
39 + accName string
40 + hasCharts bool
41 + updated bool
42 + }
43 +)
44 +
45 +func (c *accCache) put(name string) {
46 + acc, ok := (*c)[name]
47 + if !ok {
48 + acc = &accCacheEntry{accName: name}
49 + (*c)[name] = acc
50 + }
51 + acc.updated = true
52 +}
53 +
54 +type (
55 + routeCache map[uint64]*routeCacheEntry
56 + routeCacheEntry struct {
57 + rid uint64
58 + remoteId string
59 + hasCharts bool
60 + updated bool
61 + }
62 +)
63 +
64 +func (c *routeCache) put(rid uint64, remoteId string) {
65 + route, ok := (*c)[rid]
66 + if !ok {
67 + route = &routeCacheEntry{rid: rid, remoteId: remoteId}
68 + (*c)[rid] = route
69 + }
70 + route.updated = true
71 +}
72 +
73 +type (
74 + gwCache map[string]*gwCacheEntry
75 + gwCacheEntry struct {
76 + gwName string
77 + rgwName string
78 + hasCharts bool
79 + updated bool
80 + conns map[uint64]*gwConnCacheEntry
81 + }
82 + gwConnCacheEntry struct {
83 + gwName string
84 + rgwName string
85 + cid uint64
86 + hasCharts bool
87 + updated bool
88 + }
89 +)
90 +
91 +func (c *gwCache) put(gwName, rgwName string) {
92 + gw, ok := (*c)[gwName]
93 + if !ok {
94 + gw = &gwCacheEntry{gwName: gwName, rgwName: rgwName, conns: make(map[uint64]*gwConnCacheEntry)}
95 + (*c)[gwName] = gw
96 + }
97 + gw.updated = true
98 +}
99 +
100 +func (c *gwCache) putConn(gwName, rgwName string, cid uint64) {
101 + c.put(gwName, rgwName)
102 + conn, ok := (*c)[gwName].conns[cid]
103 + if !ok {
104 + conn = &gwConnCacheEntry{gwName: gwName, rgwName: rgwName, cid: cid}
105 + (*c)[gwName].conns[cid] = conn
106 + }
107 + conn.updated = true
108 +}
src/go/plugin/go.d/collector/nats/charts.go
+186 -31
@@ -4,9 +4,13 @@ package nats
4
5 import (
6 "fmt"
7 + "maps"
8 "strconv"
9 "strings"
10
11 + "golang.org/x/text/cases"
12 + "golang.org/x/text/language"
13 +
14 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/module"
15 )
16
@@ -32,6 +36,11 @@ const (
36 prioRouteTraffic
37 prioRouteMessages
38 prioRouteSubscriptions
39 +
40 + prioGatewayConnTraffic
41 + prioGatewayConnMessages
42 + prioGatewayConnSubscriptions
43 + prioGatewayConnUptime
44 )
45
46 var serverCharts = func() module.Charts {
@@ -276,29 +285,6 @@ var (
285 }
286 )
287
279 -func (c *Collector) addAccountCharts(acc string) {
280 - charts := accountChartsTmpl.Copy()
281 -
282 - for _, chart := range *charts {
283 - chart.ID = fmt.Sprintf(chart.ID, acc)
284 - chart.Labels = []module.Label{
285 - {Key: "account", Value: acc},
286 - }
287 - for _, dim := range chart.Dims {
288 - dim.ID = fmt.Sprintf(dim.ID, acc)
289 - }
290 - }
291 -
292 - if err := c.Charts().Add(*charts...); err != nil {
293 - c.Warningf("failed to add charts for account %s: %s", acc, err)
294 - }
295 -}
296 -
297 -func (c *Collector) removeAccountCharts(acc string) {
298 - px := fmt.Sprintf("account_%s_", acc)
299 - c.removeCharts(px)
300 -}
301 -
288 var routeChartsTmpl = module.Charts{
289 routeTrafficTmpl.Copy(),
290 routeMessagesTmpl.Copy(),
@@ -346,28 +332,197 @@ var (
332 }
333 )
334
349 -func (c *Collector) addRouteCharts(rid uint64, remoteId string) {
335 +var gatewayConnChartsTmpl = module.Charts{
336 + gatewayConnTrafficTmpl.Copy(),
337 + gatewayConnMessagesTmpl.Copy(),
338 + gatewayConnSubscriptionsTmpl.Copy(),
339 + gatewayConnUptime.Copy(),
340 +}
341 +
342 +var (
343 + gatewayConnTrafficTmpl = module.Chart{
344 + ID: "%s_gw_%s_cid_%d_traffic",
345 + Title: "%s Gateway Traffic",
346 + Units: "bytes/s",
347 + Fam: "gw traffic",
348 + Ctx: "nats.%s_gateway_conn_traffic",
349 + Priority: prioGatewayConnTraffic,
350 + Type: module.Area,
351 + Dims: module.Dims{
352 + {ID: "gatewayz_%s_gw_%s_cid_%d_in_bytes", Name: "in", Algo: module.Incremental},
353 + {ID: "gatewayz_%s_gw_%s_cid_%d_out_bytes", Name: "out", Mul: -1, Algo: module.Incremental},
354 + },
355 + }
356 + gatewayConnMessagesTmpl = module.Chart{
357 + ID: "%s_gw_%s_cid_%d_messages",
358 + Title: "%s Gateway Messages",
359 + Units: "messages/s",
360 + Fam: "gw traffic",
361 + Ctx: "nats.%s_gateway_conn_messages",
362 + Priority: prioGatewayConnMessages,
363 + Type: module.Line,
364 + Dims: module.Dims{
365 + {ID: "gatewayz_%s_gw_%s_cid_%d_in_msgs", Name: "in", Algo: module.Incremental},
366 + {ID: "gatewayz_%s_gw_%s_cid_%d_out_msgs", Name: "out", Mul: -1, Algo: module.Incremental},
367 + },
368 + }
369 + gatewayConnSubscriptionsTmpl = module.Chart{
370 + ID: "%s_gw_%s_cid_%d_subscriptions",
371 + Title: "%s Gateway Active Subscriptions",
372 + Units: "subscriptions",
373 + Fam: "gw subscriptions",
374 + Ctx: "nats.%s_gateway_conn_subscriptions",
375 + Priority: prioGatewayConnSubscriptions,
376 + Type: module.Line,
377 + Dims: module.Dims{
378 + {ID: "gatewayz_%s_gw_%s_cid_%d_num_subs", Name: "active"},
379 + },
380 + }
381 + gatewayConnUptime = module.Chart{
382 + ID: "%s_gw_%s_cid_%d_uptime",
383 + Title: "%s Gateway Connection Uptime",
384 + Units: "seconds",
385 + Fam: "gw uptime",
386 + Ctx: "nats.%s_gateway_conn_uptime",
387 + Priority: prioGatewayConnUptime,
388 + Dims: module.Dims{
389 + {ID: "gatewayz_%s_gw_%s_cid_%d_uptime", Name: "uptime"},
390 + },
391 + }
392 +)
393 +
394 +func (c *Collector) updateCharts() {
395 + maps.DeleteFunc(c.cache.accounts, func(_ string, acc *accCacheEntry) bool {
396 + if !acc.updated {
397 + c.removeAccountCharts(acc)
398 + return true
399 + }
400 + if !acc.hasCharts {
401 + acc.hasCharts = true
402 + c.addAccountCharts(acc)
403 + }
404 + return false
405 + })
406 + maps.DeleteFunc(c.cache.routes, func(_ uint64, route *routeCacheEntry) bool {
407 + if !route.updated {
408 + c.removeRouteCharts(route)
409 + return true
410 + }
411 + if !route.hasCharts {
412 + route.hasCharts = true
413 + c.addRouteCharts(route)
414 + }
415 + return false
416 + })
417 + maps.DeleteFunc(c.cache.inGateways, func(_ string, igw *gwCacheEntry) bool {
418 + maps.DeleteFunc(igw.conns, func(_ uint64, inConn *gwConnCacheEntry) bool {
419 + if !inConn.updated {
420 + c.removeGatewayConnCharts(inConn, true)
421 + return true
422 + }
423 + if !inConn.hasCharts {
424 + inConn.hasCharts = true
425 + c.addGatewayConnCharts(inConn, true)
426 + }
427 + return false
428 + })
429 + return false
430 + })
431 + maps.DeleteFunc(c.cache.outGateways, func(_ string, ogw *gwCacheEntry) bool {
432 + maps.DeleteFunc(ogw.conns, func(_ uint64, outConn *gwConnCacheEntry) bool {
433 + if !outConn.updated {
434 + c.removeGatewayConnCharts(outConn, false)
435 + return true
436 + }
437 + if !outConn.hasCharts {
438 + outConn.hasCharts = true
439 + c.addGatewayConnCharts(outConn, false)
440 + }
441 + return false
442 + })
443 + return false
444 + })
445 +}
446 +
447 +func (c *Collector) addAccountCharts(acc *accCacheEntry) {
448 + charts := accountChartsTmpl.Copy()
449 +
450 + for _, chart := range *charts {
451 + chart.ID = fmt.Sprintf(chart.ID, acc.accName)
452 + chart.Labels = []module.Label{
453 + {Key: "account", Value: acc.accName},
454 + }
455 + for _, dim := range chart.Dims {
456 + dim.ID = fmt.Sprintf(dim.ID, acc.accName)
457 + }
458 + }
459 +
460 + if err := c.Charts().Add(*charts...); err != nil {
461 + c.Warningf("failed to add charts for account %s: %s", acc.accName, err)
462 + }
463 +}
464 +
465 +func (c *Collector) removeAccountCharts(acc *accCacheEntry) {
466 + px := fmt.Sprintf("account_%s_", acc.accName)
467 + c.removeCharts(px)
468 +}
469 +
470 +func (c *Collector) addRouteCharts(route *routeCacheEntry) {
471 charts := routeChartsTmpl.Copy()
472
473 for _, chart := range *charts {
353 - chart.ID = fmt.Sprintf(chart.ID, rid)
474 + chart.ID = fmt.Sprintf(chart.ID, route.rid)
475 chart.Labels = []module.Label{
355 - {Key: "route_id", Value: strconv.FormatUint(rid, 10)},
356 - {Key: "remote_id", Value: remoteId},
476 + {Key: "route_id", Value: strconv.FormatUint(route.rid, 10)},
477 + {Key: "remote_id", Value: route.remoteId},
478 }
479 for _, dim := range chart.Dims {
359 - dim.ID = fmt.Sprintf(dim.ID, rid)
480 + dim.ID = fmt.Sprintf(dim.ID, route.rid)
481 }
482 }
483
484 if err := c.Charts().Add(*charts...); err != nil {
364 - c.Warningf("failed to add charts for route id %d: %s", rid, err)
485 + c.Warningf("failed to add charts for route id %d: %s", route.rid, err)
486 + }
487 +}
488 +
489 +func (c *Collector) removeRouteCharts(route *routeCacheEntry) {
490 + px := fmt.Sprintf("route_%d_", route.rid)
491 + c.removeCharts(px)
492 +}
493 +
494 +func (c *Collector) addGatewayConnCharts(gwConn *gwConnCacheEntry, isInbound bool) {
495 + direction := "outbound"
496 + if isInbound {
497 + direction = "inbound"
498 + }
499 +
500 + charts := gatewayConnChartsTmpl.Copy()
501 +
502 + for _, chart := range *charts {
503 + chart.ID = fmt.Sprintf(chart.ID, direction, gwConn.rgwName, gwConn.cid)
504 + chart.Title = fmt.Sprintf(chart.Title, cases.Title(language.English, cases.Compact).String(direction))
505 + chart.Ctx = fmt.Sprintf(chart.Ctx, direction)
506 + chart.Labels = []module.Label{
507 + {Key: "gateway", Value: gwConn.gwName},
508 + {Key: "remote_gateway", Value: gwConn.rgwName},
509 + }
510 + for _, dim := range chart.Dims {
511 + dim.ID = fmt.Sprintf(dim.ID, direction, gwConn.rgwName, gwConn.cid)
512 + }
513 }
514
515 + if err := c.Charts().Add(*charts...); err != nil {
516 + c.Warningf("failed to add charts for gateway %s %s %d: %s", direction, gwConn.rgwName, gwConn.cid, err)
517 + }
518 }
519
369 -func (c *Collector) removeRouteCharts(rid uint64) {
370 - px := fmt.Sprintf("route_%d_", rid)
520 +func (c *Collector) removeGatewayConnCharts(gwConn *gwConnCacheEntry, isInbound bool) {
521 + direction := "outbound"
522 + if isInbound {
523 + direction = "inbound"
524 + }
525 + px := fmt.Sprintf("%s_gw_%s_cid_%d_", direction, gwConn.rgwName, gwConn.cid)
526 c.removeCharts(px)
527 }
528
src/go/plugin/go.d/collector/nats/collect.go
+103 -32
@@ -5,6 +5,9 @@ package nats
5 import (
6 "fmt"
7 "net/http"
8 + "strconv"
9 + "strings"
10 + "time"
11
12 "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/metrix"
13 "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/web"
@@ -13,6 +16,8 @@ import (
16 func (c *Collector) collect() (map[string]int64, error) {
17 mx := make(map[string]int64)
18
19 + c.cache.resetUpdated()
20 +
21 if err := c.collectHealthz(mx); err != nil {
22 return nil, err
23 }
@@ -25,6 +30,11 @@ func (c *Collector) collect() (map[string]int64, error) {
30 if err := c.collectRoutez(mx); err != nil {
31 return mx, err
32 }
33 + if err := c.collectGatewayz(mx); err != nil {
34 + return mx, err
35 + }
36 +
37 + c.updateCharts()
38
39 return mx, nil
40 }
@@ -68,7 +78,8 @@ func (c *Collector) collectVarz(mx map[string]int64) error {
78 return err
79 }
80
71 - mx["varz_srv_uptime"] = int64(resp.Now.Sub(resp.Start).Seconds())
81 + uptime, _ := parseUptime(resp.Uptime)
82 + mx["varz_srv_uptime"] = int64(uptime.Seconds())
83 mx["varz_srv_in_msgs"] = resp.InMsgs
84 mx["varz_srv_out_msgs"] = resp.OutMsgs
85 mx["varz_srv_in_bytes"] = resp.InBytes
@@ -103,14 +114,8 @@ func (c *Collector) collectAccstatz(mx map[string]int64) error {
114 return err
115 }
116
106 - seen := make(map[string]bool)
107 -
117 for _, acc := range resp.AccStats {
109 - if acc.Account == "" {
110 - continue
111 - }
112 -
113 - seen[acc.Account] = true
118 + c.cache.accounts.put(acc.Account)
119
120 px := fmt.Sprintf("accstatz_acc_%s_", acc.Account)
121
@@ -125,19 +130,6 @@ func (c *Collector) collectAccstatz(mx map[string]int64) error {
130 mx[px+"sent_msgs"] = acc.Sent.Msgs
131 }
132
128 - for acc := range seen {
129 - if !c.seenAccounts[acc] {
130 - c.seenAccounts[acc] = true
131 - c.addAccountCharts(acc)
132 - }
133 - }
134 - for acc := range c.seenAccounts {
135 - if !seen[acc] {
136 - delete(c.seenAccounts, acc)
137 - c.removeAccountCharts(acc)
138 - }
139 - }
140 -
133 return nil
134 }
135
@@ -152,14 +144,8 @@ func (c *Collector) collectRoutez(mx map[string]int64) error {
144 return err
145 }
146
155 - seen := make(map[uint64]bool)
156 -
147 for _, route := range resp.Routes {
158 - seen[route.Rid] = true
159 - if !c.seenRoutes[route.Rid] {
160 - c.seenRoutes[route.Rid] = true
161 - c.addRouteCharts(route.Rid, route.RemoteID)
162 - }
148 + c.cache.routes.put(route.Rid, route.RemoteID)
149
150 px := fmt.Sprintf("routez_route_id_%d_", route.Rid)
151
@@ -170,12 +156,97 @@ func (c *Collector) collectRoutez(mx map[string]int64) error {
156 mx[px+"num_subs"] = int64(route.NumSubs)
157 }
158
173 - for rid := range c.seenRoutes {
174 - if !seen[rid] {
175 - delete(c.seenRoutes, rid)
176 - c.removeRouteCharts(rid)
159 + return nil
160 +}
161 +
162 +func (c *Collector) collectGatewayz(mx map[string]int64) error {
163 + req, err := web.NewHTTPRequestWithPath(c.RequestConfig, urlPathGatewayz)
164 + if err != nil {
165 + return err
166 + }
167 +
168 + var resp gatewayzResponse
169 + if err := web.DoHTTP(c.httpClient).RequestJSON(req, &resp); err != nil {
170 + return err
171 + }
172 +
173 + for name, ogw := range resp.OutboundGateways {
174 + c.cache.outGateways.put(resp.Name, name)
175 + c.cache.outGateways.putConn(resp.Name, name, ogw.Connection.Cid)
176 +
177 + px := fmt.Sprintf("gatewayz_outbound_gw_%s_cid_%d_", name, ogw.Connection.Cid)
178 +
179 + mx[px+"in_bytes"] = ogw.Connection.InBytes
180 + mx[px+"out_bytes"] = ogw.Connection.OutBytes
181 + mx[px+"in_msgs"] = ogw.Connection.InMsgs
182 + mx[px+"out_msgs"] = ogw.Connection.OutMsgs
183 + mx[px+"num_subs"] = int64(ogw.Connection.NumSubs)
184 + uptime, _ := parseUptime(ogw.Connection.Uptime)
185 + mx[px+"uptime"] = int64(uptime.Seconds())
186 + }
187 +
188 + for name, igws := range resp.InboundGateways {
189 + c.cache.inGateways.put(resp.Name, name)
190 + for _, igw := range igws {
191 + c.cache.inGateways.putConn(resp.Name, name, igw.Connection.Cid)
192 +
193 + px := fmt.Sprintf("gatewayz_inbound_gw_%s_cid_%d_", name, igw.Connection.Cid)
194 +
195 + mx[px+"in_bytes"] = igw.Connection.InBytes
196 + mx[px+"out_bytes"] = igw.Connection.OutBytes
197 + mx[px+"in_msgs"] = igw.Connection.InMsgs
198 + mx[px+"out_msgs"] = igw.Connection.OutMsgs
199 + mx[px+"num_subs"] = int64(igw.Connection.NumSubs)
200 + uptime, _ := parseUptime(igw.Connection.Uptime)
201 + mx[px+"uptime"] = int64(uptime.Seconds())
202 }
203 }
204
205 return nil
206 }
207 +
208 +func parseUptime(uptime string) (time.Duration, error) {
209 + // https://github.com/nats-io/nats-server/blob/v2.10.24/server/monitor.go#L1354
210 +
211 + var duration time.Duration
212 + var num strings.Builder
213 +
214 + for i := 0; i < len(uptime); i++ {
215 + ch := uptime[i]
216 + if ch >= '0' && ch <= '9' {
217 + num.WriteByte(ch)
218 + continue
219 + }
220 +
221 + if num.Len() == 0 {
222 + return 0, fmt.Errorf("invalid format: unit '%c' without number", ch)
223 + }
224 +
225 + n, err := strconv.Atoi(num.String())
226 + if err != nil {
227 + return 0, fmt.Errorf("invalid number in duration: %s", num.String())
228 + }
229 +
230 + switch ch {
231 + case 'y':
232 + duration += time.Duration(n) * 365 * 24 * time.Hour
233 + case 'd':
234 + duration += time.Duration(n) * 24 * time.Hour
235 + case 'h':
236 + duration += time.Duration(n) * time.Hour
237 + case 'm':
238 + duration += time.Duration(n) * time.Minute
239 + case 's':
240 + duration += time.Duration(n) * time.Second
241 + default:
242 + return 0, fmt.Errorf("invalid unit in duration: %c", ch)
243 + }
244 + num.Reset()
245 + }
246 +
247 + if num.Len() > 0 {
248 + return 0, fmt.Errorf("invalid format: number without unit at end")
249 + }
250 +
251 + return duration, nil
252 +}
src/go/plugin/go.d/collector/nats/collector.go
+3 -5
@@ -39,9 +39,8 @@ func New() *Collector {
39 },
40 HealthzCheck: "default",
41 },
42 - charts: serverCharts.Copy(),
43 - seenAccounts: make(map[string]bool),
44 - seenRoutes: make(map[uint64]bool),
42 + charts: serverCharts.Copy(),
43 + cache: newCache(),
44 }
45 }
46
@@ -60,8 +59,7 @@ type Collector struct {
59
60 httpClient *http.Client
61
63 - seenAccounts map[string]bool
64 - seenRoutes map[uint64]bool
62 + cache *cache
63 }
64
65 func (c *Collector) Configuration() any {
src/go/plugin/go.d/collector/nats/collector_test.go
+97 -64
@@ -24,6 +24,7 @@ var (
24 dataVer210Varz, _ = os.ReadFile("testdata/v2.10.24/varz.json")
25 dataVer210Accstatz, _ = os.ReadFile("testdata/v2.10.24/accstatz.json")
26 dataVer210Routez, _ = os.ReadFile("testdata/v2.10.24/routez.json")
27 + dataVer210Gatewayz, _ = os.ReadFile("testdata/v2.10.24/gatewayz.json")
28 )
29
30 func Test_testDataIsValid(t *testing.T) {
@@ -34,6 +35,7 @@ func Test_testDataIsValid(t *testing.T) {
35 "dataVer210Varz": dataVer210Varz,
36 "dataVer210Accstatz": dataVer210Accstatz,
37 "dataVer210Routez": dataVer210Routez,
38 + "dataVer210Gatewayz": dataVer210Gatewayz,
39 } {
40 require.NotNil(t, data, name)
41 }
@@ -131,69 +133,100 @@ func TestCollector_Collect(t *testing.T) {
133 prepare: caseOk,
134 wantNumOfCharts: len(serverCharts) +
135 len(accountChartsTmpl)*3 +
134 - len(routeChartsTmpl)*1,
136 + len(routeChartsTmpl)*1 +
137 + len(gatewayConnChartsTmpl)*5,
138 wantMetrics: map[string]int64{
136 - "accstatz_acc_$G_conns": 0,
137 - "accstatz_acc_$G_leaf_nodes": 0,
138 - "accstatz_acc_$G_num_subs": 5,
139 - "accstatz_acc_$G_received_bytes": 0,
140 - "accstatz_acc_$G_received_msgs": 0,
141 - "accstatz_acc_$G_sent_bytes": 0,
142 - "accstatz_acc_$G_sent_msgs": 0,
143 - "accstatz_acc_$G_slow_consumers": 0,
144 - "accstatz_acc_$G_total_conns": 0,
145 - "accstatz_acc_$SYS_conns": 0,
146 - "accstatz_acc_$SYS_leaf_nodes": 0,
147 - "accstatz_acc_$SYS_num_subs": 220,
148 - "accstatz_acc_$SYS_received_bytes": 0,
149 - "accstatz_acc_$SYS_received_msgs": 0,
150 - "accstatz_acc_$SYS_sent_bytes": 0,
151 - "accstatz_acc_$SYS_sent_msgs": 0,
152 - "accstatz_acc_$SYS_slow_consumers": 0,
153 - "accstatz_acc_$SYS_total_conns": 0,
154 - "accstatz_acc_default_conns": 44,
155 - "accstatz_acc_default_leaf_nodes": 0,
156 - "accstatz_acc_default_num_subs": 1133,
157 - "accstatz_acc_default_received_bytes": 62023455,
158 - "accstatz_acc_default_received_msgs": 916392,
159 - "accstatz_acc_default_sent_bytes": 529749990,
160 - "accstatz_acc_default_sent_msgs": 2546732,
161 - "accstatz_acc_default_slow_consumers": 1,
162 - "accstatz_acc_default_total_conns": 44,
163 - "routez_route_id_1_in_bytes": 4,
164 - "routez_route_id_1_in_msgs": 1,
165 - "routez_route_id_1_num_subs": 1,
166 - "routez_route_id_1_out_bytes": 4,
167 - "routez_route_id_1_out_msgs": 1,
168 - "varz_http_endpoint_/_req": 5710,
169 - "varz_http_endpoint_/accountz_req": 2201,
170 - "varz_http_endpoint_/accstatz_req": 6,
171 - "varz_http_endpoint_/connz_req": 3649,
172 - "varz_http_endpoint_/gatewayz_req": 2204,
173 - "varz_http_endpoint_/healthz_req": 3430,
174 - "varz_http_endpoint_/ipqueuesz_req": 0,
175 - "varz_http_endpoint_/jsz_req": 2958,
176 - "varz_http_endpoint_/leafz_req": 9,
177 - "varz_http_endpoint_/raftz_req": 0,
178 - "varz_http_endpoint_/routez_req": 2202,
179 - "varz_http_endpoint_/stacksz_req": 0,
180 - "varz_http_endpoint_/subsz_req": 4412,
181 - "varz_http_endpoint_/varz_req": 7114,
182 - "varz_srv_connections": 44,
183 - "varz_srv_cpu": 10,
184 - "varz_srv_healthz_status_error": 0,
185 - "varz_srv_healthz_status_ok": 1,
186 - "varz_srv_in_bytes": 62024985,
187 - "varz_srv_in_msgs": 916475,
188 - "varz_srv_mem": 95731712,
189 - "varz_srv_out_bytes": 529775656,
190 - "varz_srv_out_msgs": 2546840,
191 - "varz_srv_remotes": 0,
192 - "varz_srv_routes": 0,
193 - "varz_srv_slow_consumers": 1,
194 - "varz_srv_subscriptions": 1358,
195 - "varz_srv_total_connections": 74932,
196 - "varz_srv_uptime": 339394,
139 + "accstatz_acc_$G_conns": 0,
140 + "accstatz_acc_$G_leaf_nodes": 0,
141 + "accstatz_acc_$G_num_subs": 5,
142 + "accstatz_acc_$G_received_bytes": 0,
143 + "accstatz_acc_$G_received_msgs": 0,
144 + "accstatz_acc_$G_sent_bytes": 0,
145 + "accstatz_acc_$G_sent_msgs": 0,
146 + "accstatz_acc_$G_slow_consumers": 0,
147 + "accstatz_acc_$G_total_conns": 0,
148 + "accstatz_acc_$SYS_conns": 0,
149 + "accstatz_acc_$SYS_leaf_nodes": 0,
150 + "accstatz_acc_$SYS_num_subs": 220,
151 + "accstatz_acc_$SYS_received_bytes": 0,
152 + "accstatz_acc_$SYS_received_msgs": 0,
153 + "accstatz_acc_$SYS_sent_bytes": 0,
154 + "accstatz_acc_$SYS_sent_msgs": 0,
155 + "accstatz_acc_$SYS_slow_consumers": 0,
156 + "accstatz_acc_$SYS_total_conns": 0,
157 + "accstatz_acc_default_conns": 44,
158 + "accstatz_acc_default_leaf_nodes": 0,
159 + "accstatz_acc_default_num_subs": 1133,
160 + "accstatz_acc_default_received_bytes": 62023455,
161 + "accstatz_acc_default_received_msgs": 916392,
162 + "accstatz_acc_default_sent_bytes": 529749990,
163 + "accstatz_acc_default_sent_msgs": 2546732,
164 + "accstatz_acc_default_slow_consumers": 1,
165 + "accstatz_acc_default_total_conns": 44,
166 + "gatewayz_inbound_gw_region2_cid_9_in_bytes": 0,
167 + "gatewayz_inbound_gw_region2_cid_9_in_msgs": 0,
168 + "gatewayz_inbound_gw_region2_cid_9_num_subs": 0,
169 + "gatewayz_inbound_gw_region2_cid_9_out_bytes": 0,
170 + "gatewayz_inbound_gw_region2_cid_9_out_msgs": 0,
171 + "gatewayz_inbound_gw_region2_cid_9_uptime": 6,
172 + "gatewayz_inbound_gw_region3_cid_4_in_bytes": 0,
173 + "gatewayz_inbound_gw_region3_cid_4_in_msgs": 0,
174 + "gatewayz_inbound_gw_region3_cid_4_num_subs": 0,
175 + "gatewayz_inbound_gw_region3_cid_4_out_bytes": 0,
176 + "gatewayz_inbound_gw_region3_cid_4_out_msgs": 0,
177 + "gatewayz_inbound_gw_region3_cid_4_uptime": 6,
178 + "gatewayz_inbound_gw_region3_cid_8_in_bytes": 0,
179 + "gatewayz_inbound_gw_region3_cid_8_in_msgs": 0,
180 + "gatewayz_inbound_gw_region3_cid_8_num_subs": 0,
181 + "gatewayz_inbound_gw_region3_cid_8_out_bytes": 0,
182 + "gatewayz_inbound_gw_region3_cid_8_out_msgs": 0,
183 + "gatewayz_inbound_gw_region3_cid_8_uptime": 6,
184 + "gatewayz_outbound_gw_region2_cid_7_in_bytes": 0,
185 + "gatewayz_outbound_gw_region2_cid_7_in_msgs": 0,
186 + "gatewayz_outbound_gw_region2_cid_7_num_subs": 0,
187 + "gatewayz_outbound_gw_region2_cid_7_out_bytes": 0,
188 + "gatewayz_outbound_gw_region2_cid_7_out_msgs": 0,
189 + "gatewayz_outbound_gw_region2_cid_7_uptime": 6,
190 + "gatewayz_outbound_gw_region3_cid_5_in_bytes": 0,
191 + "gatewayz_outbound_gw_region3_cid_5_in_msgs": 0,
192 + "gatewayz_outbound_gw_region3_cid_5_num_subs": 0,
193 + "gatewayz_outbound_gw_region3_cid_5_out_bytes": 0,
194 + "gatewayz_outbound_gw_region3_cid_5_out_msgs": 0,
195 + "gatewayz_outbound_gw_region3_cid_5_uptime": 6,
196 + "routez_route_id_1_in_bytes": 4,
197 + "routez_route_id_1_in_msgs": 1,
198 + "routez_route_id_1_num_subs": 1,
199 + "routez_route_id_1_out_bytes": 4,
200 + "routez_route_id_1_out_msgs": 1,
201 + "varz_http_endpoint_/_req": 5710,
202 + "varz_http_endpoint_/accountz_req": 2201,
203 + "varz_http_endpoint_/accstatz_req": 6,
204 + "varz_http_endpoint_/connz_req": 3649,
205 + "varz_http_endpoint_/gatewayz_req": 2204,
206 + "varz_http_endpoint_/healthz_req": 3430,
207 + "varz_http_endpoint_/ipqueuesz_req": 0,
208 + "varz_http_endpoint_/jsz_req": 2958,
209 + "varz_http_endpoint_/leafz_req": 9,
210 + "varz_http_endpoint_/raftz_req": 0,
211 + "varz_http_endpoint_/routez_req": 2202,
212 + "varz_http_endpoint_/stacksz_req": 0,
213 + "varz_http_endpoint_/subsz_req": 4412,
214 + "varz_http_endpoint_/varz_req": 7114,
215 + "varz_srv_connections": 44,
216 + "varz_srv_cpu": 10,
217 + "varz_srv_healthz_status_error": 0,
218 + "varz_srv_healthz_status_ok": 1,
219 + "varz_srv_in_bytes": 62024985,
220 + "varz_srv_in_msgs": 916475,
221 + "varz_srv_mem": 95731712,
222 + "varz_srv_out_bytes": 529775656,
223 + "varz_srv_out_msgs": 2546840,
224 + "varz_srv_remotes": 0,
225 + "varz_srv_routes": 0,
226 + "varz_srv_slow_consumers": 1,
227 + "varz_srv_subscriptions": 1358,
228 + "varz_srv_total_connections": 74932,
229 + "varz_srv_uptime": 339394,
230 },
231 },
232 "fail on unexpected JSON response": {
@@ -223,8 +256,6 @@ func TestCollector_Collect(t *testing.T) {
256
257 mx := collr.Collect(context.Background())
258
226 - require.Equal(t, test.wantMetrics, mx)
227 -
259 if len(test.wantMetrics) > 0 {
260 assert.Equal(t, test.wantNumOfCharts, len(*collr.Charts()), "want charts")
261
@@ -251,6 +282,8 @@ func caseOk(t *testing.T) (*Collector, func()) {
282 _, _ = w.Write(dataVer210Accstatz)
283 case urlPathRoutez:
284 _, _ = w.Write(dataVer210Routez)
285 + case urlPathGatewayz:
286 + _, _ = w.Write(dataVer210Gatewayz)
287 default:
288 w.WriteHeader(http.StatusNotFound)
289 }
src/go/plugin/go.d/collector/nats/metadata.yaml
+80 -8
@@ -238,6 +238,18 @@ modules:
238 chart_type: line
239 dimensions:
240 - name: uptime
241 + - name: http endpoint
242 + description: These metrics refer to HTTP endpoints.
243 + labels:
244 + - name: http_endpoint
245 + description: "HTTP endpoint path."
246 + metrics:
247 + - name: nats.http_endpoint_requests
248 + description: HTTP Endpoint Requests
249 + unit: requests/s
250 + chart_type: line
251 + dimensions:
252 + - name: requests
253 - name: account
254 description: These metrics refer to [Accounts](https://docs.nats.io/running-a-nats-service/nats_admin/monitoring#account-statistics).
255 labels:
@@ -316,15 +328,75 @@ modules:
328 chart_type: line
329 dimensions:
330 - name: active
319 - - name: http endpoint
320 - description: These metrics refer to HTTP endpoints.
331 + - name: inbound gateway connection
332 + description: These metrics refer to [Inbound Gateway Connections](https://docs.nats.io/running-a-nats-service/nats_admin/monitoring#gateway-information).
333 labels:
322 - - name: http_endpoint
323 - description: "HTTP endpoint path."
334 + - name: gateway
335 + description: "The name of the local gateway."
336 + - name: remote_gateway
337 + description: "The name of the remote gateway."
338 + - name: cid
339 + description: "A unique identifier for the connection."
340 metrics:
325 - - name: nats.http_endpoint_requests
326 - description: HTTP Endpoint Requests
327 - unit: requests/s
341 + - name: nats.inbound_gateway_conn_traffic
342 + description: Inbound Gateway Traffic
343 + unit: bytes/s
344 + chart_type: area
345 + dimensions:
346 + - name: in
347 + - name: out
348 + - name: nats.inbound_gateway_conn_messages
349 + description: Inbound Gateway Messages
350 + unit: messages/s
351 chart_type: line
352 dimensions:
330 - - name: requests
353 + - name: in
354 + - name: out
355 + - name: nats.inbound_gateway_conn_subscriptions
356 + description: Inbound Gateway Active Subscriptions
357 + unit: subscriptions
358 + chart_type: line
359 + dimensions:
360 + - name: active
361 + - name: nats.inbound_gateway_conn_uptime
362 + description: Inbound Gateway Connection Uptime
363 + unit: seconds
364 + chart_type: line
365 + dimensions:
366 + - name: uptime
367 + - name: outbound gateway connection
368 + description: These metrics refer to [Outbound Gateway Connections](https://docs.nats.io/running-a-nats-service/nats_admin/monitoring#gateway-information).
369 + labels:
370 + - name: gateway
371 + description: "The name of the local gateway."
372 + - name: remote_gateway
373 + description: "The name of the remote gateway."
374 + - name: cid
375 + description: "A unique identifier for the connection."
376 + metrics:
377 + - name: nats.outbound_gateway_conn_traffic
378 + description: Outbound Gateway Traffic
379 + unit: bytes/s
380 + chart_type: area
381 + dimensions:
382 + - name: in
383 + - name: out
384 + - name: nats.outbound_gateway_conn_messages
385 + description: Outbound Gateway Messages
386 + unit: messages/s
387 + chart_type: line
388 + dimensions:
389 + - name: in
390 + - name: out
391 + - name: nats.outbound_gateway_conn_subscriptions
392 + description: Outbound Gateway Active Subscriptions
393 + unit: subscriptions
394 + chart_type: line
395 + dimensions:
396 + - name: active
397 + - name: nats.outbound_gateway_conn_uptime
398 + description: Outbound Gateway Connection Uptime
399 + unit: seconds
400 + chart_type: line
401 + dimensions:
402 + - name: uptime
src/go/plugin/go.d/collector/nats/restapi.go
+39 -12
@@ -3,8 +3,6 @@
3 package nats
4
5 import (
6 - "time"
7 -
6 "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/web"
7 )
8
@@ -19,6 +17,8 @@ const (
17 urlPathAccstatz = "/accstatz"
18 // https://docs.nats.io/running-a-nats-service/nats_admin/monitoring#route-information
19 urlPathRoutez = "/routez"
20 + // https://docs.nats.io/running-a-nats-service/nats_admin/monitoring#gateway-information
21 + urlPathGatewayz = "/gatewayz"
22 )
23
24 var (
@@ -61,8 +61,7 @@ type varzResponse struct {
61 IP string `json:"ip,omitempty"`
62 MaxConn int `json:"max_connections"`
63 MaxSubs int `json:"max_subscriptions,omitempty"`
64 - Start time.Time `json:"start"`
65 - Now time.Time `json:"now"`
64 + Uptime string `json:"uptime"`
65 Mem int64 `json:"mem"`
66 CPU float64 `json:"cpu"`
67 Connections int `json:"connections"`
@@ -80,8 +79,11 @@ type varzResponse struct {
79 }
80
81 // https://github.com/nats-io/nats-server/blob/v2.10.24/server/monitor.go#L2279
83 -type accstatzResponse struct {
84 - AccStats []struct {
82 +type (
83 + accstatzResponse struct {
84 + AccStats []accountInfo `json:"account_statz"`
85 + }
86 + accountInfo struct {
87 Account string `json:"acc"`
88 Conns int `json:"conns"`
89 TotalConns int `json:"total_conns"`
@@ -96,12 +98,15 @@ type accstatzResponse struct {
98 Bytes int64 `json:"bytes"`
99 } `json:"received"`
100 SlowConsumers int64 `json:"slow_consumers"`
99 - } `json:"account_statz"`
100 -}
101 + }
102 +)
103
104 // https://github.com/nats-io/nats-server/blob/v2.10.24/server/monitor.go#L752
103 -type routezResponse struct {
104 - Routes []struct {
105 +type (
106 + routezResponse struct {
107 + Routes []routeInfo `json:"routes"`
108 + }
109 + routeInfo struct {
110 Rid uint64 `json:"rid"`
111 RemoteID string `json:"remote_id"`
112 InMsgs int64 `json:"in_msgs"`
@@ -109,5 +114,27 @@ type routezResponse struct {
114 InBytes int64 `json:"in_bytes"`
115 OutBytes int64 `json:"out_bytes"`
116 NumSubs uint32 `json:"subscriptions"`
112 - } `json:"routes"`
113 -}
117 + }
118 +)
119 +
120 +type (
121 + // https://github.com/nats-io/nats-server/blob/v2.10.24/server/monitor.go#L1875
122 + gatewayzResponse struct {
123 + Name string `json:"name"`
124 + OutboundGateways map[string]*remoteGatewayInfo `json:"outbound_gateways"`
125 + InboundGateways map[string][]*remoteGatewayInfo `json:"inbound_gateways"`
126 + }
127 + remoteGatewayInfo struct {
128 + IsConfigured bool `json:"configured"`
129 + Connection connectionInfo `json:"connection"`
130 + }
131 + connectionInfo struct {
132 + Cid uint64 `json:"cid"`
133 + Uptime string `json:"uptime"`
134 + InMsgs int64 `json:"in_msgs"`
135 + OutMsgs int64 `json:"out_msgs"`
136 + InBytes int64 `json:"in_bytes"`
137 + OutBytes int64 `json:"out_bytes"`
138 + NumSubs uint32 `json:"subscriptions"`
139 + }
140 +)
src/go/plugin/go.d/collector/nats/testdata/v2.10.24/gatewayz.json new
+110
@@ -0,0 +1,110 @@
1 +{
2 + "server_id": "NANVBOU62MDUWTXWRQ5KH3PSMYNCHCEUHQV3TW3YH7WZLS7FMJE6END6",
3 + "now": "2019-07-24T18:02:55.597398-06:00",
4 + "name": "region1",
5 + "host": "2601:283:4601:1350:1895:efda:2010:95a1",
6 + "port": 4501,
7 + "outbound_gateways": {
8 + "region2": {
9 + "configured": true,
10 + "connection": {
11 + "cid": 7,
12 + "ip": "127.0.0.1",
13 + "port": 5500,
14 + "start": "2019-07-24T18:02:48.765621-06:00",
15 + "last_activity": "2019-07-24T18:02:48.765621-06:00",
16 + "uptime": "6s",
17 + "idle": "6s",
18 + "pending_bytes": 0,
19 + "in_msgs": 0,
20 + "out_msgs": 0,
21 + "in_bytes": 0,
22 + "out_bytes": 0,
23 + "subscriptions": 0,
24 + "name": "NCXBIYWT7MV7OAQTCR4QTKBN3X3HDFGSFWTURTCQ22ZZB6NKKJPO7MN4"
25 + }
26 + },
27 + "region3": {
28 + "configured": true,
29 + "connection": {
30 + "cid": 5,
31 + "ip": "::1",
32 + "port": 6500,
33 + "start": "2019-07-24T18:02:48.764685-06:00",
34 + "last_activity": "2019-07-24T18:02:48.764685-06:00",
35 + "uptime": "6s",
36 + "idle": "6s",
37 + "pending_bytes": 0,
38 + "in_msgs": 0,
39 + "out_msgs": 0,
40 + "in_bytes": 0,
41 + "out_bytes": 0,
42 + "subscriptions": 0,
43 + "name": "NCVS7Q65WX3FGIL2YQRLI77CE6MQRWO2Y453HYVLNMBMTVLOKMPW7R6K"
44 + }
45 + }
46 + },
47 + "inbound_gateways": {
48 + "region2": [
49 + {
50 + "configured": false,
51 + "connection": {
52 + "cid": 9,
53 + "ip": "::1",
54 + "port": 52029,
55 + "start": "2019-07-24T18:02:48.76677-06:00",
56 + "last_activity": "2019-07-24T18:02:48.767096-06:00",
57 + "uptime": "6s",
58 + "idle": "6s",
59 + "pending_bytes": 0,
60 + "in_msgs": 0,
61 + "out_msgs": 0,
62 + "in_bytes": 0,
63 + "out_bytes": 0,
64 + "subscriptions": 0,
65 + "name": "NCXBIYWT7MV7OAQTCR4QTKBN3X3HDFGSFWTURTCQ22ZZB6NKKJPO7MN4"
66 + }
67 + }
68 + ],
69 + "region3": [
70 + {
71 + "configured": false,
72 + "connection": {
73 + "cid": 4,
74 + "ip": "::1",
75 + "port": 52025,
76 + "start": "2019-07-24T18:02:48.764577-06:00",
77 + "last_activity": "2019-07-24T18:02:48.764994-06:00",
78 + "uptime": "6s",
79 + "idle": "6s",
80 + "pending_bytes": 0,
81 + "in_msgs": 0,
82 + "out_msgs": 0,
83 + "in_bytes": 0,
84 + "out_bytes": 0,
85 + "subscriptions": 0,
86 + "name": "NCVS7Q65WX3FGIL2YQRLI77CE6MQRWO2Y453HYVLNMBMTVLOKMPW7R6K"
87 + }
88 + },
89 + {
90 + "configured": false,
91 + "connection": {
92 + "cid": 8,
93 + "ip": "127.0.0.1",
94 + "port": 52026,
95 + "start": "2019-07-24T18:02:48.766173-06:00",
96 + "last_activity": "2019-07-24T18:02:48.766999-06:00",
97 + "uptime": "6s",
98 + "idle": "6s",
99 + "pending_bytes": 0,
100 + "in_msgs": 0,
101 + "out_msgs": 0,
102 + "in_bytes": 0,
103 + "out_bytes": 0,
104 + "subscriptions": 0,
105 + "name": "NCKCYK5LE3VVGOJQ66F65KA27UFPCLBPX4N4YOPOXO3KHGMW24USPCKN"
106 + }
107 + }
108 + ]
109 + }
110 +}