@cryptotaxi247 / netdata-1 / commits / 5158928b9

improvement(go.d/nats): add leafz metrics (#19282)

Ilya Mashchenko committed Dec 24, 2024 at 13:29 UTC 5158928b9e3c6726e50675a374d9c3b37bbaddfb
7 files changed +289 -31
src/go/plugin/go.d/collector/nats/cache.go
+54 -24
@@ -2,12 +2,17 @@
2
3 package nats
4
5 +import (
6 + "fmt"
7 +)
8 +
9 func newCache() *cache {
10 return &cache{
11 accounts: make(accCache),
12 routes: make(routeCache),
13 inGateways: make(gwCache),
14 outGateways: make(gwCache),
15 + leafs: make(leafCache),
16 }
17 }
18
@@ -16,6 +21,7 @@ type cache struct {
21 routes routeCache
22 inGateways gwCache
23 outGateways gwCache
24 + leafs leafCache
25 }
26
27 func (c *cache) resetUpdated() {
@@ -36,17 +42,18 @@ func (c *cache) resetUpdated() {
42 type (
43 accCache map[string]*accCacheEntry
44 accCacheEntry struct {
39 - accName string
45 hasCharts bool
46 updated bool
47 +
48 + accName string
49 }
50 )
51
45 -func (c *accCache) put(name string) {
46 - acc, ok := (*c)[name]
52 +func (c *accCache) put(ai accountInfo) {
53 + acc, ok := (*c)[ai.Account]
54 if !ok {
48 - acc = &accCacheEntry{accName: name}
49 - (*c)[name] = acc
55 + acc = &accCacheEntry{accName: ai.Account}
56 + (*c)[ai.Account] = acc
57 }
58 acc.updated = true
59 }
@@ -54,18 +61,19 @@ func (c *accCache) put(name string) {
61 type (
62 routeCache map[uint64]*routeCacheEntry
63 routeCacheEntry struct {
57 - rid uint64
58 - remoteId string
64 hasCharts bool
65 updated bool
66 +
67 + rid uint64
68 + remoteId string
69 }
70 )
71
64 -func (c *routeCache) put(rid uint64, remoteId string) {
65 - route, ok := (*c)[rid]
72 +func (c *routeCache) put(ri routeInfo) {
73 + route, ok := (*c)[ri.Rid]
74 if !ok {
67 - route = &routeCacheEntry{rid: rid, remoteId: remoteId}
68 - (*c)[rid] = route
75 + route = &routeCacheEntry{rid: ri.Rid, remoteId: ri.RemoteID}
76 + (*c)[ri.Rid] = route
77 }
78 route.updated = true
79 }
@@ -73,36 +81,58 @@ func (c *routeCache) put(rid uint64, remoteId string) {
81 type (
82 gwCache map[string]*gwCacheEntry
83 gwCacheEntry struct {
76 - gwName string
77 - rgwName string
84 hasCharts bool
85 updated bool
80 - conns map[uint64]*gwConnCacheEntry
86 +
87 + gwName string
88 + rgwName string
89 + conns map[uint64]*gwConnCacheEntry
90 }
91 gwConnCacheEntry struct {
83 - gwName string
84 - rgwName string
85 - cid uint64
92 hasCharts bool
93 updated bool
94 +
95 + gwName string
96 + rgwName string
97 + cid uint64
98 }
99 )
100
91 -func (c *gwCache) put(gwName, rgwName string) {
101 +func (c *gwCache) put(gwName, rgwName string, rgi *remoteGatewayInfo) {
102 gw, ok := (*c)[gwName]
103 if !ok {
104 gw = &gwCacheEntry{gwName: gwName, rgwName: rgwName, conns: make(map[uint64]*gwConnCacheEntry)}
105 (*c)[gwName] = gw
106 }
107 gw.updated = true
98 -}
108
100 -func (c *gwCache) putConn(gwName, rgwName string, cid uint64) {
101 - c.put(gwName, rgwName)
102 - conn, ok := (*c)[gwName].conns[cid]
109 + conn, ok := gw.conns[rgi.Connection.Cid]
110 if !ok {
104 - conn = &gwConnCacheEntry{gwName: gwName, rgwName: rgwName, cid: cid}
105 - (*c)[gwName].conns[cid] = conn
111 + conn = &gwConnCacheEntry{gwName: gwName, rgwName: rgwName, cid: rgi.Connection.Cid}
112 + gw.conns[rgi.Connection.Cid] = conn
113 }
114 conn.updated = true
115 }
116 +
117 +type (
118 + leafCache map[string]*leafCacheEntry
119 + leafCacheEntry struct {
120 + hasCharts bool
121 + updated bool
122 +
123 + leafName string
124 + account string
125 + ip string
126 + port int
127 + }
128 +)
129 +
130 +func (c *leafCache) put(li leafInfo) {
131 + key := fmt.Sprintf("%s_%s_%s_%d", li.Name, li.Account, li.IP, li.Port)
132 + leaf, ok := (*c)[key]
133 + if !ok {
134 + leaf = &leafCacheEntry{leafName: li.Name, account: li.Account, ip: li.IP, port: li.Port}
135 + (*c)[key] = leaf
136 + }
137 + leaf.updated = true
138 +}
src/go/plugin/go.d/collector/nats/charts.go
+109
@@ -41,6 +41,11 @@ const (
41 prioGatewayConnMessages
42 prioGatewayConnSubscriptions
43 prioGatewayConnUptime
44 +
45 + prioLeafConnTraffic
46 + prioLeafConnMessages
47 + prioLeafConnSubscriptions
48 + prioLeafRTT
49 )
50
51 var serverCharts = func() module.Charts {
@@ -391,6 +396,65 @@ var (
396 }
397 )
398
399 +var leafConnChartsTmpl = module.Charts{
400 + leafConnTrafficTmpl.Copy(),
401 + leafConnMessagesTmpl.Copy(),
402 + leafConnSubscriptionsTmpl.Copy(),
403 + leafConnRTT.Copy(),
404 +}
405 +
406 +var (
407 + leafConnTrafficTmpl = module.Chart{
408 + ID: "leaf_node_conn_%s_%s_%s_%d_traffic",
409 + Title: "Leaf Node Connection Traffic",
410 + Units: "bytes/s",
411 + Fam: "leaf traffic",
412 + Ctx: "nats.leaf_node_conn_traffic",
413 + Priority: prioLeafConnTraffic,
414 + Type: module.Area,
415 + Dims: module.Dims{
416 + {ID: "leafz_leaf_%s_%s_%s_%d_in_bytes", Name: "in", Algo: module.Incremental},
417 + {ID: "leafz_leaf_%s_%s_%s_%d_out_bytes", Name: "out", Mul: -1, Algo: module.Incremental},
418 + },
419 + }
420 + leafConnMessagesTmpl = module.Chart{
421 + ID: "leaf_node_conn_%s_%s_%s_%d_messages",
422 + Title: "Leaf Node Connection Messages",
423 + Units: "messages/s",
424 + Fam: "leaf traffic",
425 + Ctx: "nats.leaf_node_conn_messages",
426 + Priority: prioLeafConnMessages,
427 + Type: module.Line,
428 + Dims: module.Dims{
429 + {ID: "leafz_leaf_%s_%s_%s_%d_in_msgs", Name: "in", Algo: module.Incremental},
430 + {ID: "leafz_leaf_%s_%s_%s_%d_out_msgs", Name: "out", Mul: -1, Algo: module.Incremental},
431 + },
432 + }
433 + leafConnSubscriptionsTmpl = module.Chart{
434 + ID: "leaf_node_conn_%s_%s_%s_%d_subscriptions",
435 + Title: "Leaf Node Connection Active Subscriptions",
436 + Units: "subscriptions",
437 + Fam: "leaf subscriptions",
438 + Ctx: "nats.leaf_node_conn_subscriptions",
439 + Priority: prioLeafConnSubscriptions,
440 + Type: module.Line,
441 + Dims: module.Dims{
442 + {ID: "leafz_leaf_%s_%s_%s_%d_num_subs", Name: "active"},
443 + },
444 + }
445 + leafConnRTT = module.Chart{
446 + ID: "leaf_node_conn_%s_%s_%s_%d_rtt",
447 + Title: "Leaf Node Connection RTT",
448 + Units: "microseconds",
449 + Fam: "leaf rtt",
450 + Ctx: "nats.leaf_node_conn_rtt",
451 + Priority: prioLeafRTT,
452 + Dims: module.Dims{
453 + {ID: "leafz_leaf_%s_%s_%s_%d_rtt", Name: "rtt"},
454 + },
455 + }
456 +)
457 +
458 func (c *Collector) updateCharts() {
459 c.onceAddSrvCharts.Do(c.addServerCharts)
460
@@ -444,6 +508,17 @@ func (c *Collector) updateCharts() {
508 })
509 return false
510 })
511 + maps.DeleteFunc(c.cache.leafs, func(_ string, leaf *leafCacheEntry) bool {
512 + if !leaf.updated {
513 + c.removeLeafCharts(leaf)
514 + return true
515 + }
516 + if !leaf.hasCharts {
517 + leaf.hasCharts = true
518 + c.addLeafCharts(leaf)
519 + }
520 + return false
521 + })
522 }
523
524 func (c *Collector) addServerCharts() {
@@ -545,6 +620,35 @@ func (c *Collector) removeGatewayConnCharts(gwConn *gwConnCacheEntry, isInbound
620 c.removeCharts(px)
621 }
622
623 +func (c *Collector) addLeafCharts(leaf *leafCacheEntry) {
624 + charts := leafConnChartsTmpl.Copy()
625 +
626 + for _, chart := range *charts {
627 + chart.ID = fmt.Sprintf(chart.ID, leaf.leafName, leaf.account, leaf.ip, leaf.port)
628 + chart.ID = cleanChartID(chart.ID)
629 + chart.Labels = []module.Label{
630 + {Key: "server_id", Value: c.srvMeta.id},
631 + {Key: "remote_name", Value: leaf.leafName},
632 + {Key: "account", Value: leaf.account},
633 + {Key: "ip", Value: leaf.ip},
634 + {Key: "port", Value: strconv.Itoa(leaf.port)},
635 + }
636 + for _, dim := range chart.Dims {
637 + dim.ID = fmt.Sprintf(dim.ID, leaf.leafName, leaf.account, leaf.ip, leaf.port)
638 + }
639 + }
640 +
641 + if err := c.Charts().Add(*charts...); err != nil {
642 + c.Warningf("failed to add charts for leaf %s: %s", leaf.leafName, err)
643 + }
644 +}
645 +
646 +func (c *Collector) removeLeafCharts(leaf *leafCacheEntry) {
647 + px := fmt.Sprintf("leaf_node_conn_%s_%s_%s_%d_", leaf.leafName, leaf.account, leaf.ip, leaf.port)
648 + cleanChartID(px)
649 + c.removeCharts(px)
650 +}
651 +
652 func (c *Collector) removeCharts(prefix string) {
653 for _, chart := range *c.Charts() {
654 if strings.HasPrefix(chart.ID, prefix) {
@@ -553,3 +657,8 @@ func (c *Collector) removeCharts(prefix string) {
657 }
658 }
659 }
660 +
661 +func cleanChartID(id string) string {
662 + r := strings.NewReplacer(".", "_", " ", "_")
663 + return strings.ToLower(r.Replace(id))
664 +}
src/go/plugin/go.d/collector/nats/collect.go
+34 -6
@@ -42,6 +42,9 @@ func (c *Collector) collect() (map[string]int64, error) {
42 if err := c.collectGatewayz(mx); err != nil {
43 return mx, err
44 }
45 + if err := c.collectLeafz(mx); err != nil {
46 + return mx, err
47 + }
48
49 c.updateCharts()
50
@@ -142,7 +145,7 @@ func (c *Collector) collectAccstatz(mx map[string]int64) error {
145 }
146
147 for _, acc := range resp.AccStats {
145 - c.cache.accounts.put(acc.Account)
148 + c.cache.accounts.put(acc)
149
150 px := fmt.Sprintf("accstatz_acc_%s_", acc.Account)
151
@@ -172,7 +175,7 @@ func (c *Collector) collectRoutez(mx map[string]int64) error {
175 }
176
177 for _, route := range resp.Routes {
175 - c.cache.routes.put(route.Rid, route.RemoteID)
178 + c.cache.routes.put(route)
179
180 px := fmt.Sprintf("routez_route_id_%d_", route.Rid)
181
@@ -198,8 +201,7 @@ func (c *Collector) collectGatewayz(mx map[string]int64) error {
201 }
202
203 for name, ogw := range resp.OutboundGateways {
201 - c.cache.outGateways.put(resp.Name, name)
202 - c.cache.outGateways.putConn(resp.Name, name, ogw.Connection.Cid)
204 + c.cache.outGateways.put(resp.Name, name, ogw)
205
206 px := fmt.Sprintf("gatewayz_outbound_gw_%s_cid_%d_", name, ogw.Connection.Cid)
207
@@ -213,9 +215,8 @@ func (c *Collector) collectGatewayz(mx map[string]int64) error {
215 }
216
217 for name, igws := range resp.InboundGateways {
216 - c.cache.inGateways.put(resp.Name, name)
218 for _, igw := range igws {
218 - c.cache.inGateways.putConn(resp.Name, name, igw.Connection.Cid)
219 + c.cache.inGateways.put(resp.Name, name, igw)
220
221 px := fmt.Sprintf("gatewayz_inbound_gw_%s_cid_%d_", name, igw.Connection.Cid)
222
@@ -232,6 +233,33 @@ func (c *Collector) collectGatewayz(mx map[string]int64) error {
233 return nil
234 }
235
236 +func (c *Collector) collectLeafz(mx map[string]int64) error {
237 + req, err := web.NewHTTPRequestWithPath(c.RequestConfig, urlPathLeafz)
238 + if err != nil {
239 + return err
240 + }
241 +
242 + var resp leafzResponse
243 + if err := web.DoHTTP(c.httpClient).RequestJSON(req, &resp); err != nil {
244 + return err
245 + }
246 +
247 + for _, leaf := range resp.Leafs {
248 + c.cache.leafs.put(leaf)
249 + px := fmt.Sprintf("leafz_leaf_%s_%s_%s_%d_", leaf.Name, leaf.Account, leaf.IP, leaf.Port)
250 +
251 + mx[px+"in_bytes"] = leaf.InBytes
252 + mx[px+"out_bytes"] = leaf.OutBytes
253 + mx[px+"in_msgs"] = leaf.InMsgs
254 + mx[px+"out_msgs"] = leaf.OutMsgs
255 + mx[px+"num_subs"] = int64(leaf.NumSubs)
256 + rtt, _ := time.ParseDuration(leaf.RTT)
257 + mx[px+"rtt"] = rtt.Microseconds()
258 + }
259 +
260 + return nil
261 +}
262 +
263 func parseUptime(uptime string) (time.Duration, error) {
264 // https://github.com/nats-io/nats-server/blob/v2.10.24/server/monitor.go#L1354
265
src/go/plugin/go.d/collector/nats/collector_test.go
+12 -1
@@ -25,6 +25,7 @@ var (
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 + dataVer210Leafz, _ = os.ReadFile("testdata/v2.10.24/leafz.json")
29 )
30
31 func Test_testDataIsValid(t *testing.T) {
@@ -36,6 +37,7 @@ func Test_testDataIsValid(t *testing.T) {
37 "dataVer210Accstatz": dataVer210Accstatz,
38 "dataVer210Routez": dataVer210Routez,
39 "dataVer210Gatewayz": dataVer210Gatewayz,
40 + "dataVer210Leafz": dataVer210Leafz,
41 } {
42 require.NotNil(t, data, name)
43 }
@@ -134,7 +136,8 @@ func TestCollector_Collect(t *testing.T) {
136 wantNumOfCharts: len(serverCharts) +
137 len(accountChartsTmpl)*3 +
138 len(routeChartsTmpl)*1 +
137 - len(gatewayConnChartsTmpl)*5,
139 + len(gatewayConnChartsTmpl)*5 +
140 + len(leafConnChartsTmpl)*1,
141 wantMetrics: map[string]int64{
142 "accstatz_acc_$G_conns": 0,
143 "accstatz_acc_$G_leaf_nodes": 0,
@@ -193,6 +196,12 @@ func TestCollector_Collect(t *testing.T) {
196 "gatewayz_outbound_gw_region3_cid_5_out_bytes": 0,
197 "gatewayz_outbound_gw_region3_cid_5_out_msgs": 0,
198 "gatewayz_outbound_gw_region3_cid_5_uptime": 6,
199 + "leafz_leaf__$G_127.0.0.1_6223_in_bytes": 0,
200 + "leafz_leaf__$G_127.0.0.1_6223_in_msgs": 0,
201 + "leafz_leaf__$G_127.0.0.1_6223_num_subs": 1,
202 + "leafz_leaf__$G_127.0.0.1_6223_out_bytes": 1280000,
203 + "leafz_leaf__$G_127.0.0.1_6223_out_msgs": 10000,
204 + "leafz_leaf__$G_127.0.0.1_6223_rtt": 0,
205 "routez_route_id_1_in_bytes": 4,
206 "routez_route_id_1_in_msgs": 1,
207 "routez_route_id_1_num_subs": 1,
@@ -284,6 +293,8 @@ func caseOk(t *testing.T) (*Collector, func()) {
293 _, _ = w.Write(dataVer210Routez)
294 case urlPathGatewayz:
295 _, _ = w.Write(dataVer210Gatewayz)
296 + case urlPathLeafz:
297 + _, _ = w.Write(dataVer210Leafz)
298 default:
299 w.WriteHeader(http.StatusNotFound)
300 }
src/go/plugin/go.d/collector/nats/metadata.yaml
+38
@@ -412,3 +412,41 @@ modules:
412 chart_type: line
413 dimensions:
414 - name: uptime
415 + - name: leaf node connection
416 + description: These metrics refer to [Leaf Node Connections](https://docs.nats.io/running-a-nats-service/nats_admin/monitoring#leaf-node-information).
417 + labels:
418 + - name: remote_name
419 + description: "Unique identifier of the remote leaf node server, either its configured name or automatically assigned ID."
420 + - name: account
421 + description: "Name of the associated account."
422 + - name: ip
423 + description: "IP address of the remote server."
424 + - name: port
425 + description: "Port used for the connection to the remote server."
426 + metrics:
427 + - name: nats.leaf_node_conn_traffic
428 + description: Leaf Node Connection Traffic
429 + unit: bytes/s
430 + chart_type: area
431 + dimensions:
432 + - name: in
433 + - name: out
434 + - name: nats.leaf_node_conn_messages
435 + description: Leaf Node Connection Messages
436 + unit: messages/s
437 + chart_type: line
438 + dimensions:
439 + - name: in
440 + - name: out
441 + - name: nats.leaf_node_conn_subscriptions
442 + description: Leaf Node Connection Active Subscriptions
443 + unit: subscriptions
444 + chart_type: line
445 + dimensions:
446 + - name: active
447 + - name: nats.leaf_node_conn_rtt
448 + description: Leaf Node Connection RTT
449 + unit: microseconds
450 + chart_type: line
451 + dimensions:
452 + - name: rtt
src/go/plugin/go.d/collector/nats/restapi.go
+21
@@ -19,6 +19,8 @@ const (
19 urlPathRoutez = "/routez"
20 // https://docs.nats.io/running-a-nats-service/nats_admin/monitoring#gateway-information
21 urlPathGatewayz = "/gatewayz"
22 + // https://docs.nats.io/running-a-nats-service/nats_admin/monitoring#leaf-node-information
23 + urlPathLeafz = "/leafz"
24 )
25
26 var (
@@ -138,3 +140,22 @@ type (
140 NumSubs uint32 `json:"subscriptions"`
141 }
142 )
143 +
144 +type (
145 + // https://github.com/nats-io/nats-server/blob/v2.10.24/server/monitor.go#L2163
146 + leafzResponse struct {
147 + Leafs []leafInfo `json:"leafs"`
148 + }
149 + leafInfo struct {
150 + Name string `json:"name"` // remote server name or id
151 + Account string `json:"account"`
152 + IP string `json:"ip"`
153 + Port int `json:"port"`
154 + RTT string `json:"rtt,omitempty"`
155 + InMsgs int64 `json:"in_msgs"`
156 + OutMsgs int64 `json:"out_msgs"`
157 + InBytes int64 `json:"in_bytes"`
158 + OutBytes int64 `json:"out_bytes"`
159 + NumSubs uint32 `json:"subscriptions"`
160 + }
161 +)
src/go/plugin/go.d/collector/nats/testdata/v2.10.24/leafz.json new
+21
@@ -0,0 +1,21 @@
1 +{
2 + "server_id": "NC2FJCRMPBE5RI5OSRN7TKUCWQONCKNXHKJXCJIDVSAZ6727M7MQFVT3",
3 + "now": "2019-08-27T09:07:05.841132-06:00",
4 + "leafnodes": 1,
5 + "leafs": [
6 + {
7 + "account": "$G",
8 + "ip": "127.0.0.1",
9 + "port": 6223,
10 + "rtt": "200µs",
11 + "in_msgs": 0,
12 + "out_msgs": 10000,
13 + "in_bytes": 0,
14 + "out_bytes": 1280000,
15 + "subscriptions": 1,
16 + "subscriptions_list": [
17 + "foo"
18 + ]
19 + }
20 + ]
21 +}