master
go 484 lines 11.3 KB
Raw
1 // Package jmx provides a typed adapter around the WebSphere JMX helper bridge.
2 // SPDX-License-Identifier: GPL-3.0-or-later
3
4 //go:build cgo
5
6 package jmx
7
8 import (
9 "context"
10 "encoding/json"
11 "errors"
12 "fmt"
13 "strconv"
14 "strings"
15 "time"
16
17 "github.com/netdata/netdata/go/plugins/plugin/ibm.d/protocols/jmxbridge"
18 )
19
20 // NewClient constructs a WebSphere JMX protocol client.
21 func NewClient(cfg Config, logger jmxbridge.Logger, opts ...Option) (*Client, error) {
22 if logger == nil {
23 return nil, errors.New("websphere jmx protocol: logger is required")
24 }
25
26 trimmedURL := strings.TrimSpace(cfg.JMXURL)
27 if trimmedURL == "" {
28 return nil, errors.New("websphere jmx protocol: jmx_url is required")
29 }
30 cfg.JMXURL = trimmedURL
31
32 if cfg.InitTimeout <= 0 {
33 cfg.InitTimeout = 30 * time.Second
34 }
35 if cfg.CommandTimeout <= 0 {
36 cfg.CommandTimeout = 5 * time.Second
37 }
38 if cfg.ShutdownDelay < 0 {
39 cfg.ShutdownDelay = 0
40 }
41
42 client := &Client{
43 cfg: cfg,
44 logger: logger,
45 jarData: helperJar,
46 jarName: helperJarName,
47 }
48
49 for _, opt := range opts {
50 opt(client)
51 }
52
53 if client.bridge == nil {
54 bridgeCfg := jmxbridge.Config{
55 JavaExecPath: cfg.JavaExecPath,
56 JarData: client.jarData,
57 JarFileName: client.jarName,
58 }
59 bridge, err := jmxbridge.NewClient(bridgeCfg, logger)
60 if err != nil {
61 return nil, fmt.Errorf("websphere jmx protocol: creating bridge failed: %w", err)
62 }
63 client.bridge = bridge
64 }
65
66 return client, nil
67 }
68
69 // Start launches the helper process and performs the INIT handshake.
70 func (c *Client) Start(ctx context.Context) error {
71 c.mu.Lock()
72 defer c.mu.Unlock()
73
74 if c.started {
75 return nil
76 }
77
78 startCtx, cancel := context.WithTimeout(ctx, c.cfg.InitTimeout)
79 defer cancel()
80
81 cmd := jmxbridge.Command{
82 "command": "INIT",
83 "protocol_version": protocolVersion,
84 "jmx_url": c.cfg.JMXURL,
85 }
86 if c.cfg.JMXUsername != "" {
87 cmd["jmx_username"] = c.cfg.JMXUsername
88 }
89 if c.cfg.JMXPassword != "" {
90 cmd["jmx_password"] = c.cfg.JMXPassword
91 }
92 if c.cfg.JMXClasspath != "" {
93 cmd["jmx_classpath"] = c.cfg.JMXClasspath
94 }
95
96 if err := c.bridge.Start(startCtx, cmd); err != nil {
97 return fmt.Errorf("websphere jmx protocol: helper init failed: %w", err)
98 }
99
100 c.started = true
101 return nil
102 }
103
104 // Shutdown stops the helper process.
105 func (c *Client) Shutdown() {
106 c.mu.Lock()
107 defer c.mu.Unlock()
108
109 if !c.started {
110 return
111 }
112
113 if c.cfg.ShutdownDelay > 0 {
114 time.Sleep(c.cfg.ShutdownDelay)
115 }
116
117 c.bridge.Shutdown()
118 c.started = false
119 }
120
121 // FetchJVM requests JVM metrics from the helper.
122 func (c *Client) FetchJVM(ctx context.Context) (*JVMStats, error) {
123 payload, err := c.send(ctx, jmxbridge.Command{
124 "command": "SCRAPE",
125 "target": "JVM",
126 })
127 if err != nil {
128 return nil, err
129 }
130
131 stats := &JVMStats{}
132
133 heap := mapValue(payload, "heap")
134 stats.Heap.Used = floatValue(heap, "used")
135 stats.Heap.Committed = floatValue(heap, "committed")
136 stats.Heap.Max = floatValue(heap, "max")
137
138 nonheap := mapValue(payload, "nonheap")
139 stats.NonHeap.Used = floatValue(nonheap, "used")
140 stats.NonHeap.Committed = floatValue(nonheap, "committed")
141
142 gc := mapValue(payload, "gc")
143 stats.GC.Count = floatValue(gc, "count")
144 stats.GC.Time = floatValue(gc, "time")
145
146 threads := mapValue(payload, "threads")
147 stats.Threads.Count = floatValue(threads, "count")
148 stats.Threads.Daemon = floatValue(threads, "daemon")
149 stats.Threads.Peak = floatValue(threads, "peak")
150 stats.Threads.Started = floatValue(threads, "totalStarted")
151
152 classes := mapValue(payload, "classes")
153 stats.Classes.Loaded = floatValue(classes, "loaded")
154 stats.Classes.Unloaded = floatValue(classes, "unloaded")
155
156 cpu := mapValue(payload, "cpu")
157 stats.CPU.ProcessUsage = floatValue(cpu, "processCpuUsage")
158
159 stats.Uptime = floatValue(payload, "uptime")
160
161 return stats, nil
162 }
163
164 // FetchThreadPools retrieves thread pool metrics from the helper.
165 func (c *Client) FetchThreadPools(ctx context.Context, maxItems int) ([]ThreadPool, error) {
166 cmd := jmxbridge.Command{
167 "command": "SCRAPE",
168 "target": "THREADPOOLS",
169 }
170 if maxItems > 0 {
171 cmd["max_items"] = maxItems
172 }
173
174 payload, err := c.send(ctx, cmd)
175 if err != nil {
176 return nil, err
177 }
178
179 var pools []ThreadPool
180 items, ok := payload["threadPools"].([]any)
181 if !ok {
182 return pools, nil
183 }
184
185 for _, item := range items {
186 poolMap, ok := item.(map[string]any)
187 if !ok {
188 continue
189 }
190 name := stringValue(poolMap, "name")
191 if name == "" {
192 continue
193 }
194
195 pools = append(pools, ThreadPool{
196 Name: name,
197 PoolSize: floatValue(poolMap, "poolSize"),
198 ActiveCount: floatValue(poolMap, "activeCount"),
199 MaximumPoolSize: floatValue(poolMap, "maximumPoolSize"),
200 })
201 }
202
203 return pools, nil
204 }
205
206 // FetchJDBCPools retrieves JDBC pool statistics from the helper.
207 func (c *Client) FetchJDBCPools(ctx context.Context, maxItems int) ([]JDBCPool, error) {
208 cmd := jmxbridge.Command{
209 "command": "SCRAPE",
210 "target": "JDBC",
211 }
212 if maxItems > 0 {
213 cmd["max_items"] = maxItems
214 }
215
216 payload, err := c.send(ctx, cmd)
217 if err != nil {
218 return nil, err
219 }
220
221 var pools []JDBCPool
222 items, ok := payload["jdbcPools"].([]any)
223 if !ok {
224 return pools, nil
225 }
226
227 for _, item := range items {
228 poolMap, ok := item.(map[string]any)
229 if !ok {
230 continue
231 }
232 name := stringValue(poolMap, "name")
233 if name == "" {
234 continue
235 }
236
237 pools = append(pools, JDBCPool{
238 Name: name,
239 PoolSize: floatValue(poolMap, "poolSize"),
240 NumConnectionsUsed: floatValue(poolMap, "numConnectionsUsed"),
241 NumConnectionsFree: floatValue(poolMap, "numConnectionsFree"),
242 AvgWaitTime: floatValue(poolMap, "avgWaitTime"),
243 AvgInUseTime: floatValue(poolMap, "avgInUseTime"),
244 NumConnectionsCreated: floatValue(poolMap, "numConnectionsCreated"),
245 NumConnectionsDestroyed: floatValue(poolMap, "numConnectionsDestroyed"),
246 WaitingThreadCount: floatValue(poolMap, "waitingThreadCount"),
247 })
248 }
249
250 return pools, nil
251 }
252
253 // FetchJCAPools retrieves JCA pool statistics from the helper.
254 func (c *Client) FetchJCAPools(ctx context.Context, maxItems int) ([]JCAPool, error) {
255 cmd := jmxbridge.Command{
256 "command": "SCRAPE",
257 "target": "JCA",
258 }
259 if maxItems > 0 {
260 cmd["max_items"] = maxItems
261 }
262
263 payload, err := c.send(ctx, cmd)
264 if err != nil {
265 return nil, err
266 }
267
268 var pools []JCAPool
269 items, ok := payload["jcaPools"].([]any)
270 if !ok {
271 return pools, nil
272 }
273
274 for _, item := range items {
275 poolMap, ok := item.(map[string]any)
276 if !ok {
277 continue
278 }
279 name := stringValue(poolMap, "name")
280 if name == "" {
281 continue
282 }
283
284 pools = append(pools, JCAPool{
285 Name: name,
286 PoolSize: floatValue(poolMap, "poolSize"),
287 NumConnectionsUsed: floatValue(poolMap, "numConnectionsUsed"),
288 NumConnectionsFree: floatValue(poolMap, "numConnectionsFree"),
289 AvgWaitTime: floatValue(poolMap, "avgWaitTime"),
290 AvgInUseTime: floatValue(poolMap, "avgInUseTime"),
291 NumConnectionsCreated: floatValue(poolMap, "numConnectionsCreated"),
292 NumConnectionsDestroyed: floatValue(poolMap, "numConnectionsDestroyed"),
293 WaitingThreadCount: floatValue(poolMap, "waitingThreadCount"),
294 })
295 }
296
297 return pools, nil
298 }
299
300 // FetchJMSDestinations retrieves JMS metrics from the helper.
301 func (c *Client) FetchJMSDestinations(ctx context.Context, maxItems int) ([]JMSDestination, error) {
302 cmd := jmxbridge.Command{
303 "command": "SCRAPE",
304 "target": "JMS",
305 }
306 if maxItems > 0 {
307 cmd["max_items"] = maxItems
308 }
309
310 payload, err := c.send(ctx, cmd)
311 if err != nil {
312 return nil, err
313 }
314
315 var dests []JMSDestination
316 items, ok := payload["jmsDestinations"].([]any)
317 if !ok {
318 return dests, nil
319 }
320
321 for _, item := range items {
322 destMap, ok := item.(map[string]any)
323 if !ok {
324 continue
325 }
326 name := stringValue(destMap, "name")
327 if name == "" {
328 continue
329 }
330
331 dests = append(dests, JMSDestination{
332 Name: name,
333 Type: stringValue(destMap, "type"),
334 MessagesCurrentCount: floatValue(destMap, "messagesCurrentCount"),
335 MessagesPendingCount: floatValue(destMap, "messagesPendingCount"),
336 MessagesAddedCount: floatValue(destMap, "messagesAddedCount"),
337 ConsumerCount: floatValue(destMap, "consumerCount"),
338 })
339 }
340
341 return dests, nil
342 }
343
344 // FetchApplications retrieves web application metrics from the helper.
345 func (c *Client) FetchApplications(ctx context.Context, maxItems int, includeSessions, includeTransactions bool) ([]ApplicationMetric, error) {
346 cmd := jmxbridge.Command{
347 "command": "SCRAPE",
348 "target": "APPLICATIONS",
349 }
350 if maxItems > 0 {
351 cmd["max_items"] = maxItems
352 }
353 cmd["collect_options"] = map[string]bool{
354 "sessions": includeSessions,
355 "transactions": includeTransactions,
356 }
357
358 payload, err := c.send(ctx, cmd)
359 if err != nil {
360 return nil, err
361 }
362
363 var metrics []ApplicationMetric
364 items, ok := payload["applications"].([]any)
365 if !ok {
366 return metrics, nil
367 }
368
369 for _, item := range items {
370 appMap, ok := item.(map[string]any)
371 if !ok {
372 continue
373 }
374 name := stringValue(appMap, "name")
375 if name == "" {
376 continue
377 }
378
379 metric := ApplicationMetric{
380 Name: name,
381 Module: stringValue(appMap, "module"),
382 Requests: floatValue(appMap, "requestCount"),
383 ResponseTime: floatValue(appMap, "averageResponseTime"),
384 ActiveSessions: floatValue(appMap, "activeSessions"),
385 LiveSessions: floatValue(appMap, "liveSessions"),
386 SessionCreates: floatValue(appMap, "sessionCreates"),
387 SessionInvalidates: floatValue(appMap, "sessionInvalidates"),
388 TransactionsCommitted: floatValue(appMap, "transactionsCommitted"),
389 TransactionsRolledback: floatValue(appMap, "transactionsRolledBack"),
390 }
391
392 metrics = append(metrics, metric)
393 }
394
395 return metrics, nil
396 }
397
398 func (c *Client) send(ctx context.Context, cmd jmxbridge.Command) (map[string]any, error) {
399 if !c.started {
400 return nil, errors.New("websphere jmx protocol: client not started")
401 }
402
403 cmdCtx, cancel := context.WithTimeout(ctx, c.cfg.CommandTimeout)
404 defer cancel()
405
406 resp, err := c.bridge.Send(cmdCtx, cmd)
407 if err != nil {
408 if resp != nil {
409 return nil, fmt.Errorf("websphere jmx protocol: command failed: %s", resp.Message)
410 }
411 return nil, fmt.Errorf("websphere jmx protocol: command failed: %w", err)
412 }
413
414 if resp == nil {
415 return nil, errors.New("websphere jmx protocol: empty response")
416 }
417
418 return resp.Data, nil
419 }
420
421 func mapValue(m map[string]any, key string) map[string]any {
422 if m == nil {
423 return map[string]any{}
424 }
425 val, _ := m[key].(map[string]any)
426 if val == nil {
427 return map[string]any{}
428 }
429 return val
430 }
431
432 func stringValue(m map[string]any, key string) string {
433 if m == nil {
434 return ""
435 }
436 if v, ok := m[key].(string); ok {
437 return v
438 }
439 return ""
440 }
441
442 func floatValue(m map[string]any, key string) float64 {
443 if m == nil {
444 return 0
445 }
446 return toFloat(m[key])
447 }
448
449 func toFloat(v any) float64 {
450 switch value := v.(type) {
451 case nil:
452 return 0
453 case float64:
454 return value
455 case float32:
456 return float64(value)
457 case int:
458 return float64(value)
459 case int64:
460 return float64(value)
461 case int32:
462 return float64(value)
463 case uint:
464 return float64(value)
465 case uint64:
466 return float64(value)
467 case uint32:
468 return float64(value)
469 case string:
470 parsed, err := strconv.ParseFloat(value, 64)
471 if err == nil {
472 return parsed
473 }
474 return 0
475 case json.Number:
476 parsed, err := value.Float64()
477 if err == nil {
478 return parsed
479 }
480 return 0
481 default:
482 return 0
483 }
484 }