master
go 1,142 lines 34 KB
Raw
1 //go:build cgo
2
3 package as400
4
5 // SPDX-License-Identifier: GPL-3.0-or-later
6
7 import (
8 "context"
9 "errors"
10 "fmt"
11 "strings"
12 "sync"
13 "time"
14
15 "github.com/netdata/netdata/go/plugins/pkg/matcher"
16 "github.com/netdata/netdata/go/plugins/pkg/stm"
17 "github.com/netdata/netdata/go/plugins/plugin/ibm.d/framework"
18 "github.com/netdata/netdata/go/plugins/plugin/ibm.d/modules/as400/contexts"
19 "github.com/netdata/netdata/go/plugins/plugin/ibm.d/pkg/dbdriver"
20 as400proto "github.com/netdata/netdata/go/plugins/plugin/ibm.d/protocols/as400"
21 )
22
23 // Collector implements the IBM i module on top of the ibm.d framework.
24 type Collector struct {
25 framework.Collector
26
27 Config `yaml:",inline" json:",inline"`
28
29 client *as400proto.Client
30
31 // Per-iteration metrics
32 mx *metricsData
33
34 fastQueryLatencyCounters map[string]int64
35 batchLatencyValues contexts.ObservabilityQueryLatencyBatchValues
36 batchLatencyValid bool
37
38 // Metadata caches (reset every iteration)
39 disks map[string]*diskMetrics
40 subsystems map[string]*subsystemMetrics
41 jobQueues map[string]*jobQueueMetrics
42 messageQueues map[string]*messageQueueMetrics
43 outputQueues map[string]*outputQueueMetrics
44 tempStorageNamed map[string]*tempStorageMetrics
45 activeJobs map[string]*activeJobMetrics
46 networkInterfaces map[string]*networkInterfaceMetrics
47 httpServers map[string]*httpServerMetrics
48 planCache map[string]*planCacheMetrics
49
50 // Selectors
51 diskSelector matcher.Matcher
52 subsystemSelector matcher.Matcher
53
54 slow struct {
55 client *as400proto.Client
56 cancel context.CancelFunc
57 wg sync.WaitGroup
58 config slowPathConfig
59 cache slowCache
60 }
61
62 batch struct {
63 client *as400proto.Client
64 cancel context.CancelFunc
65 wg sync.WaitGroup
66 config batchPathConfig
67 cache batchCache
68 }
69
70 // System identity
71 systemName string
72 serialNumber string
73 model string
74 osVersion string
75 technologyRefresh string
76 versionMajor int
77 versionRelease int
78 versionMod int
79
80 // Feature flags and logging guards
81 disabled map[string]bool
82 errorLogged map[string]bool
83 muErrorLog sync.Mutex
84
85 // Cardinality guards to avoid repeated expensive counts
86 diskCardinality cardinalityGuard
87 networkInterfacesCardinality cardinalityGuard
88 httpServersCardinality cardinalityGuard
89
90 dump *dumpContext
91 groups []collectionGroup
92
93 messageQueueTargets []queueTarget
94 jobQueueTargets []queueTarget
95 outputQueueTargets []queueTarget
96 activeJobTargets []activeJobTarget
97
98 // CPU collection state for delta-based calculation
99 cpuCollectionMethod string // "total_cpu_time" or "elapsed_cpu_used"
100 prevTotalCPUTime int64 // Previous TOTAL_CPU_TIME value (nanoseconds)
101 prevElapsedTime int64 // Previous ELAPSED_TIME value (seconds)
102 prevElapsedCPUProduct int64 // Previous ELAPSED_CPU_USED * ELAPSED_TIME product
103 hasCPUBaseline bool // Whether we have a previous measurement
104
105 once sync.Once
106 }
107
108 func (c *Collector) initOnce() {
109 c.once.Do(func() {
110 c.disabled = make(map[string]bool)
111 c.errorLogged = make(map[string]bool)
112 c.mx = &metricsData{}
113 c.resetInstanceCaches()
114 c.initGroups()
115 })
116 }
117
118 func (c *Collector) resetInstanceCaches() {
119 c.disks = make(map[string]*diskMetrics)
120 c.subsystems = make(map[string]*subsystemMetrics)
121 c.jobQueues = make(map[string]*jobQueueMetrics)
122 c.messageQueues = make(map[string]*messageQueueMetrics)
123 c.outputQueues = make(map[string]*outputQueueMetrics)
124 c.tempStorageNamed = make(map[string]*tempStorageMetrics)
125 c.activeJobs = make(map[string]*activeJobMetrics)
126 c.networkInterfaces = make(map[string]*networkInterfaceMetrics)
127 c.httpServers = make(map[string]*httpServerMetrics)
128 c.planCache = make(map[string]*planCacheMetrics)
129 c.mx.disks = make(map[string]diskInstanceMetrics)
130 c.mx.subsystems = make(map[string]subsystemInstanceMetrics)
131 c.mx.jobQueues = make(map[string]jobQueueInstanceMetrics)
132 c.mx.messageQueues = make(map[string]messageQueueInstanceMetrics)
133 c.mx.outputQueues = make(map[string]outputQueueInstanceMetrics)
134 c.mx.tempStorageNamed = make(map[string]tempStorageInstanceMetrics)
135 c.mx.activeJobs = make(map[string]activeJobInstanceMetrics)
136 c.mx.networkInterfaces = make(map[string]networkInterfaceInstanceMetrics)
137 c.mx.httpServers = make(map[string]httpServerInstanceMetrics)
138 c.mx.planCache = make(map[string]planCacheInstanceMetrics)
139 }
140
141 func (c *Collector) prepareIterationState() {
142 c.mx = &metricsData{
143 systemActivity: systemActivityMetrics{},
144 }
145 c.resetInstanceCaches()
146 c.diskCardinality.Configure(c.MaxDisks)
147 c.networkInterfacesCardinality.Configure(networkInterfaceLimit)
148 c.httpServersCardinality.Configure(httpServerLimit)
149 }
150
151 func (c *Collector) initGroups() {
152 if c.groups != nil {
153 return
154 }
155 c.groups = []collectionGroup{
156 &systemGroup{c},
157 &diskGroup{c},
158 &subsystemGroup{c},
159 &jobQueueGroup{c},
160 &messageQueueGroup{c},
161 &outputQueueGroup{c},
162 &activeJobGroup{c},
163 &networkInterfaceGroup{c},
164 &systemActivityGroup{c},
165 &httpServerGroup{c},
166 &planCacheGroup{c},
167 }
168 }
169
170 // CollectOnce implements framework.CollectorImpl.
171 func (c *Collector) CollectOnce() error {
172 c.initOnce()
173
174 ctx := context.Background()
175 if err := c.client.Connect(ctx); err != nil {
176 return err
177 }
178 if err := c.client.Ping(ctx); err != nil {
179 _ = c.client.Close()
180 if err := c.client.Connect(ctx); err != nil {
181 return err
182 }
183 if err := c.client.Ping(ctx); err != nil {
184 return err
185 }
186 }
187
188 if err := c.collect(ctx); err != nil {
189 return err
190 }
191 if c.dump != nil {
192 c.dump.recordMetrics(c.snapshotMetrics())
193 }
194
195 // Populate contexts from metric struct maps
196 c.exportSystemMetrics()
197 c.exportDiskMetrics()
198 c.exportSubsystemMetrics()
199 c.exportJobQueueMetrics()
200 c.exportMessageQueueMetrics()
201 c.exportOutputQueueMetrics()
202 c.exportQueueTotalsMetrics()
203 c.exportTempStorageMetrics()
204 c.exportActiveJobMetrics()
205 c.exportNetworkInterfaceMetrics()
206 c.exportSystemActivityMetrics()
207 c.exportHTTPServerMetrics()
208 c.exportPlanCacheMetrics()
209 c.exportQueryLatencyMetrics()
210 c.applyGlobalLabels()
211
212 return nil
213 }
214
215 func (c *Collector) snapshotMetrics() map[string]int64 {
216 metrics := make(map[string]int64)
217
218 for k, v := range stm.ToMap(c.mx) {
219 if isSystemMetric(k) {
220 metrics[k] = v
221 }
222 }
223
224 for unit, values := range c.mx.disks {
225 clean := cleanName(unit)
226 for k, v := range stm.ToMap(values) {
227 metrics[fmt.Sprintf("disk_%s_%s", clean, k)] = v
228 }
229 if values.SSDLifeRemaining >= 0 {
230 metrics[fmt.Sprintf("disk_%s_ssd_life_remaining", clean)] = values.SSDLifeRemaining
231 }
232 if values.SSDPowerOnDays >= 0 {
233 metrics[fmt.Sprintf("disk_%s_ssd_power_on_days", clean)] = values.SSDPowerOnDays
234 }
235 }
236
237 for name, values := range c.mx.subsystems {
238 clean := cleanName(name)
239 for k, v := range stm.ToMap(values) {
240 metrics[fmt.Sprintf("subsystem_%s_%s", clean, k)] = v
241 }
242 }
243
244 for name, values := range c.mx.jobQueues {
245 clean := cleanName(name)
246 for k, v := range stm.ToMap(values) {
247 metrics[fmt.Sprintf("jobqueue_%s_%s", clean, k)] = v
248 }
249 }
250
251 for key, values := range c.mx.messageQueues {
252 clean := cleanName(key)
253 for k, v := range stm.ToMap(values) {
254 metrics[fmt.Sprintf("message_queue_%s_%s", clean, k)] = v
255 }
256 }
257
258 for key, values := range c.mx.outputQueues {
259 clean := cleanName(key)
260 for k, v := range stm.ToMap(values) {
261 metrics[fmt.Sprintf("output_queue_%s_%s", clean, k)] = v
262 }
263 }
264
265 for bucket, values := range c.mx.tempStorageNamed {
266 clean := cleanName(bucket)
267 for k, v := range stm.ToMap(values) {
268 metrics[fmt.Sprintf("tempstorage_%s_%s", clean, k)] = v
269 }
270 }
271
272 for jobName, values := range c.mx.activeJobs {
273 clean := cleanName(jobName)
274 for k, v := range stm.ToMap(values) {
275 metrics[fmt.Sprintf("activejob_%s_%s", clean, k)] = v
276 }
277 }
278
279 for name, values := range c.mx.networkInterfaces {
280 clean := cleanName(name)
281 for k, v := range stm.ToMap(values) {
282 metrics[fmt.Sprintf("netintf_%s_%s", clean, k)] = v
283 }
284 }
285
286 for id, values := range c.mx.httpServers {
287 clean := cleanName(id)
288 for k, v := range stm.ToMap(values) {
289 metrics[fmt.Sprintf("httpserver_%s_%s", clean, k)] = v
290 }
291 }
292
293 for heading, values := range c.mx.planCache {
294 clean := cleanName(heading)
295 for k, v := range stm.ToMap(values) {
296 metrics[fmt.Sprintf("plan_cache_%s_%s", clean, k)] = v
297 }
298 }
299
300 for k, v := range stm.ToMap(c.mx.systemActivity) {
301 metrics[fmt.Sprintf("system_activity_%s", k)] = v
302 }
303
304 return metrics
305 }
306
307 func isSystemMetric(key string) bool {
308 systemMetrics := map[string]bool{
309 "cpu_percentage": true,
310 "configured_cpus": true,
311 "current_cpu_capacity": true,
312 "main_storage_size": true,
313 "current_temporary_storage": true,
314 "maximum_temporary_storage_used": true,
315 "total_jobs_in_system": true,
316 "active_jobs_in_system": true,
317 "interactive_jobs_in_system": true,
318 "batch_jobs_running": true,
319 "job_queue_length": true,
320 "system_asp_used": true,
321 "system_asp_storage": true,
322 "total_auxiliary_storage": true,
323 "active_threads_in_system": true,
324 "threads_per_processor": true,
325 "machine_pool_size": true,
326 "base_pool_size": true,
327 "interactive_pool_size": true,
328 "spool_pool_size": true,
329 "machine_pool_defined_size": true,
330 "machine_pool_reserved_size": true,
331 "base_pool_defined_size": true,
332 "base_pool_reserved_size": true,
333 "machine_pool_threads": true,
334 "machine_pool_max_threads": true,
335 "base_pool_threads": true,
336 "base_pool_max_threads": true,
337 "remote_connections": true,
338 "total_connections": true,
339 "listen_connections": true,
340 "closewait_connections": true,
341 "temp_storage_current_total": true,
342 "temp_storage_peak_total": true,
343 "disk_busy_percentage": true,
344 "system_activity_average_cpu_rate": true,
345 "system_activity_average_cpu_utilization": true,
346 "system_activity_minimum_cpu_utilization": true,
347 "system_activity_maximum_cpu_utilization": true,
348 }
349
350 return systemMetrics[key]
351 }
352
353 func (c *Collector) exportSystemMetrics() {
354 labels := contexts.EmptyLabels{}
355
356 contexts.System.CPUUtilization.Set(c.State, labels, contexts.SystemCPUUtilizationValues{
357 Utilization: c.mx.CPUPercentage,
358 })
359
360 contexts.System.CPUEntitledUtilization.Set(c.State, labels, contexts.SystemCPUEntitledUtilizationValues{
361 Utilization: c.mx.EntitledCPUPercentage,
362 })
363
364 contexts.System.CPUDetails.Set(c.State, labels, contexts.SystemCPUDetailsValues{
365 Configured: c.mx.ConfiguredCPUs,
366 })
367
368 contexts.System.CPUCapacity.Set(c.State, labels, contexts.SystemCPUCapacityValues{
369 Capacity: c.mx.CurrentCPUCapacity,
370 })
371
372 contexts.System.TotalJobs.Set(c.State, labels, contexts.SystemTotalJobsValues{
373 Total: c.mx.TotalJobsInSystem,
374 })
375
376 contexts.System.ActiveJobsByType.Set(c.State, labels, contexts.SystemActiveJobsByTypeValues{
377 Batch: c.mx.BatchJobsRunning,
378 Interactive: c.mx.InteractiveJobsInSystem,
379 Active: c.mx.ActiveJobsInSystem,
380 })
381
382 contexts.System.JobQueueLength.Set(c.State, labels, contexts.SystemJobQueueLengthValues{
383 Waiting: c.mx.JobQueueLength,
384 })
385
386 contexts.System.MainStorageSize.Set(c.State, labels, contexts.SystemMainStorageSizeValues{
387 Total: c.mx.MainStorageSize,
388 })
389
390 contexts.System.TemporaryStorage.Set(c.State, labels, contexts.SystemTemporaryStorageValues{
391 Current: c.mx.CurrentTemporaryStorage,
392 Maximum: c.mx.MaximumTemporaryStorageUsed,
393 })
394
395 contexts.System.MemoryPoolUsage.Set(c.State, labels, contexts.SystemMemoryPoolUsageValues{
396 Machine: c.mx.MachinePoolSize,
397 Base: c.mx.BasePoolSize,
398 Interactive: c.mx.InteractivePoolSize,
399 Spool: c.mx.SpoolPoolSize,
400 })
401
402 contexts.System.MemoryPoolDefined.Set(c.State, labels, contexts.SystemMemoryPoolDefinedValues{
403 Machine: c.mx.MachinePoolDefinedSize,
404 Base: c.mx.BasePoolDefinedSize,
405 })
406
407 contexts.System.MemoryPoolReserved.Set(c.State, labels, contexts.SystemMemoryPoolReservedValues{
408 Machine: c.mx.MachinePoolReservedSize,
409 Base: c.mx.BasePoolReservedSize,
410 })
411
412 contexts.System.MemoryPoolThreads.Set(c.State, labels, contexts.SystemMemoryPoolThreadsValues{
413 Machine: c.mx.MachinePoolThreads,
414 Base: c.mx.BasePoolThreads,
415 })
416
417 contexts.System.MemoryPoolMaxThreads.Set(c.State, labels, contexts.SystemMemoryPoolMaxThreadsValues{
418 Machine: c.mx.MachinePoolMaxThreads,
419 Base: c.mx.BasePoolMaxThreads,
420 })
421
422 avgDiskBusy := c.mx.DiskBusyPercentage
423 if avgDiskBusy == 0 {
424 var (
425 sum int64
426 count int64
427 )
428 for _, values := range c.mx.disks {
429 sum += values.BusyPercent
430 count++
431 }
432 if count > 0 {
433 avgDiskBusy = sum / count
434 }
435 }
436
437 if avgDiskBusy != c.mx.DiskBusyPercentage {
438 c.mx.DiskBusyPercentage = avgDiskBusy
439 }
440
441 contexts.System.DiskBusyAverage.Set(c.State, labels, contexts.SystemDiskBusyAverageValues{
442 Busy: avgDiskBusy,
443 })
444
445 contexts.System.SystemASPUsage.Set(c.State, labels, contexts.SystemSystemASPUsageValues{
446 Used: c.mx.SystemASPUsed,
447 })
448
449 contexts.System.SystemASPStorage.Set(c.State, labels, contexts.SystemSystemASPStorageValues{
450 Total: c.mx.SystemASPStorage,
451 })
452
453 contexts.System.TotalAuxiliaryStorage.Set(c.State, labels, contexts.SystemTotalAuxiliaryStorageValues{
454 Total: c.mx.TotalAuxiliaryStorage,
455 })
456
457 contexts.System.SystemThreads.Set(c.State, labels, contexts.SystemSystemThreadsValues{
458 Active: c.mx.ActiveThreadsInSystem,
459 Per_processor: c.mx.ThreadsPerProcessor,
460 })
461
462 contexts.System.NetworkConnections.Set(c.State, labels, contexts.SystemNetworkConnectionsValues{
463 Remote: c.mx.RemoteConnections,
464 Total: c.mx.TotalConnections,
465 })
466
467 contexts.System.NetworkConnectionStates.Set(c.State, labels, contexts.SystemNetworkConnectionStatesValues{
468 Listen: c.mx.ListenConnections,
469 Close_wait: c.mx.CloseWaitConnections,
470 })
471
472 contexts.System.TempStorageTotal.Set(c.State, labels, contexts.SystemTempStorageTotalValues{
473 Current: c.mx.TempStorageCurrentTotal,
474 Peak: c.mx.TempStoragePeakTotal,
475 })
476
477 if c.mx.systemActivity.AverageCPURate != 0 || c.mx.systemActivity.AverageCPUUtilization != 0 {
478 contexts.System.SystemActivityCPURate.Set(c.State, labels, contexts.SystemSystemActivityCPURateValues{
479 Average: c.mx.systemActivity.AverageCPURate,
480 })
481 contexts.System.SystemActivityCPUUtilization.Set(c.State, labels, contexts.SystemSystemActivityCPUUtilizationValues{
482 Average: c.mx.systemActivity.AverageCPUUtilization,
483 Minimum: c.mx.systemActivity.MinimumCPUUtilization,
484 Maximum: c.mx.systemActivity.MaximumCPUUtilization,
485 })
486 }
487 }
488
489 func (c *Collector) exportDiskMetrics() {
490 for unit, values := range c.mx.disks {
491 meta := c.disks[unit]
492 diskUnit := unit
493 diskType := ""
494 diskModel := ""
495 hardwareStatus := ""
496 diskSerial := ""
497 if meta != nil {
498 if meta.unit != "" {
499 diskUnit = meta.unit
500 }
501 diskType = meta.typeField
502 diskModel = meta.diskModel
503 hardwareStatus = meta.hardwareStatus
504 diskSerial = meta.serialNumber
505 }
506
507 labels := contexts.DiskLabels{
508 Disk_unit: diskUnit,
509 Disk_type: diskType,
510 Disk_model: diskModel,
511 Hardware_status: hardwareStatus,
512 Disk_serial_number: diskSerial,
513 }
514
515 contexts.Disk.Busy.Set(c.State, labels, contexts.DiskBusyValues{
516 Busy: values.BusyPercent,
517 })
518
519 contexts.Disk.IORequests.Set(c.State, labels, contexts.DiskIORequestsValues{
520 Read: values.ReadRequests,
521 Write: values.WriteRequests,
522 })
523
524 contexts.Disk.SpaceUsage.Set(c.State, labels, contexts.DiskSpaceUsageValues{
525 Used: values.PercentUsed,
526 })
527
528 contexts.Disk.Capacity.Set(c.State, labels, contexts.DiskCapacityValues{
529 Available: values.AvailableGB,
530 Used: values.UsedGB,
531 })
532
533 contexts.Disk.Blocks.Set(c.State, labels, contexts.DiskBlocksValues{
534 Read: values.BlocksRead,
535 Write: values.BlocksWritten,
536 })
537
538 if values.SSDLifeRemaining > 0 {
539 contexts.Disk.SSDHealth.Set(c.State, labels, contexts.DiskSSDHealthValues{
540 Life_remaining: values.SSDLifeRemaining,
541 })
542 }
543
544 if values.SSDPowerOnDays > 0 {
545 contexts.Disk.SSDPowerOn.Set(c.State, labels, contexts.DiskSSDPowerOnValues{
546 Power_on_days: values.SSDPowerOnDays,
547 })
548 }
549 }
550 }
551
552 func (c *Collector) exportSubsystemMetrics() {
553 for name, values := range c.mx.subsystems {
554 meta := c.subsystems[name]
555 subsystemName := name
556 library := ""
557 status := "ACTIVE"
558 if meta != nil {
559 if meta.name != "" {
560 subsystemName = meta.name
561 }
562 library = meta.library
563 if meta.status != "" {
564 status = meta.status
565 }
566 }
567 labels := contexts.SubsystemLabels{
568 Subsystem: subsystemName,
569 Library: library,
570 Status: status,
571 }
572 if interval := c.slowPathIntervalSeconds(); interval > 0 {
573 contexts.Subsystem.Jobs.SetUpdateEvery(c.State, labels, interval)
574 }
575 contexts.Subsystem.Jobs.Set(c.State, labels, contexts.SubsystemJobsValues{
576 Active: values.CurrentActiveJobs,
577 Maximum: values.MaximumActiveJobs,
578 })
579 }
580 }
581
582 func (c *Collector) exportJobQueueMetrics() {
583 for key, values := range c.mx.jobQueues {
584 meta := c.jobQueues[key]
585 queueName := key
586 library := ""
587 status := "RELEASED"
588 if meta != nil {
589 if meta.name != "" {
590 queueName = meta.name
591 }
592 library = meta.library
593 if meta.status != "" {
594 status = meta.status
595 }
596 }
597 labels := contexts.JobQueueLabels{
598 Job_queue: queueName,
599 Library: library,
600 Status: status,
601 }
602 if interval := c.slowPathIntervalSeconds(); interval > 0 {
603 contexts.JobQueue.Length.SetUpdateEvery(c.State, labels, interval)
604 }
605 contexts.JobQueue.Length.Set(c.State, labels, contexts.JobQueueLengthValues{
606 Jobs: values.NumberOfJobs,
607 })
608 }
609 }
610
611 func (c *Collector) exportTempStorageMetrics() {
612 for name, values := range c.mx.tempStorageNamed {
613 labels := contexts.TempStorageBucketLabels{Bucket: name}
614 contexts.TempStorageBucket.Usage.Set(c.State, labels, contexts.TempStorageBucketUsageValues{
615 Current: values.CurrentSize,
616 Peak: values.PeakSize,
617 })
618 }
619 }
620
621 func (c *Collector) exportMessageQueueMetrics() {
622 for key, values := range c.mx.messageQueues {
623 meta := c.messageQueues[key]
624 library := ""
625 queue := key
626 if meta != nil {
627 library = meta.library
628 if meta.name != "" {
629 queue = meta.name
630 }
631 }
632 labels := contexts.MessageQueueLabels{
633 Library: library,
634 Queue: queue,
635 }
636 if interval := c.slowPathIntervalSeconds(); interval > 0 {
637 contexts.MessageQueue.Messages.SetUpdateEvery(c.State, labels, interval)
638 contexts.MessageQueue.Severity.SetUpdateEvery(c.State, labels, interval)
639 }
640 contexts.MessageQueue.Messages.Set(c.State, labels, contexts.MessageQueueMessagesValues{
641 Total: values.Total,
642 Informational: values.Informational,
643 Inquiry: values.Inquiry,
644 Diagnostic: values.Diagnostic,
645 Escape: values.Escape,
646 Notify: values.Notify,
647 Sender_copy: values.SenderCopy,
648 })
649 contexts.MessageQueue.Severity.Set(c.State, labels, contexts.MessageQueueSeverityValues{
650 Max: values.MaxSeverity,
651 })
652 }
653 }
654
655 func (c *Collector) exportOutputQueueMetrics() {
656 for key, values := range c.mx.outputQueues {
657 meta := c.outputQueues[key]
658 library := ""
659 queue := key
660 status := ""
661 if meta != nil {
662 library = meta.library
663 status = meta.status
664 if meta.name != "" {
665 queue = meta.name
666 }
667 }
668 labels := contexts.OutputQueueLabels{
669 Library: library,
670 Queue: queue,
671 Status: status,
672 }
673 if interval := c.slowPathIntervalSeconds(); interval > 0 {
674 contexts.OutputQueue.Files.SetUpdateEvery(c.State, labels, interval)
675 contexts.OutputQueue.Writers.SetUpdateEvery(c.State, labels, interval)
676 contexts.OutputQueue.Status.SetUpdateEvery(c.State, labels, interval)
677 }
678 contexts.OutputQueue.Files.Set(c.State, labels, contexts.OutputQueueFilesValues{
679 Files: values.Files,
680 })
681 contexts.OutputQueue.Writers.Set(c.State, labels, contexts.OutputQueueWritersValues{
682 Writers: values.Writers,
683 })
684 contexts.OutputQueue.Status.Set(c.State, labels, contexts.OutputQueueStatusValues{
685 Released: values.Released,
686 })
687 }
688 }
689
690 func (c *Collector) exportQueueTotalsMetrics() {
691 if !c.batchPathActive() {
692 return
693 }
694
695 snapshot := c.batch.cache.getTotals()
696 if snapshot.timestamp.IsZero() && len(snapshot.queues) == 0 && len(snapshot.items) == 0 && snapshot.err == nil {
697 return
698 }
699
700 interval := c.batchPathIntervalSeconds()
701 types := []struct {
702 queueType string
703 itemType string
704 enabled bool
705 }{
706 {"message_queue", "message", c.CollectMessageQueueTotals.IsEnabled()},
707 {"job_queue", "job", c.CollectJobQueueTotals.IsEnabled()},
708 {"output_queue", "spooled_file", c.CollectOutputQueueTotals.IsEnabled()},
709 }
710
711 for _, entry := range types {
712 if !entry.enabled {
713 continue
714 }
715
716 labels := contexts.QueueOverviewLabels{
717 Queue_type: entry.queueType,
718 Item_type: entry.itemType,
719 }
720 if interval > 0 {
721 contexts.QueueOverview.Count.SetUpdateEvery(c.State, labels, interval)
722 contexts.QueueOverview.Items.SetUpdateEvery(c.State, labels, interval)
723 }
724 contexts.QueueOverview.Count.Set(c.State, labels, contexts.QueueOverviewCountValues{
725 Queues: snapshot.queues[entry.queueType],
726 })
727 contexts.QueueOverview.Items.Set(c.State, labels, contexts.QueueOverviewItemsValues{
728 Items: snapshot.items[entry.queueType],
729 })
730 }
731 }
732
733 func (c *Collector) exportActiveJobMetrics() {
734 for jobName, values := range c.mx.activeJobs {
735 meta := c.activeJobs[jobName]
736 jobNameLabel := jobName
737 jobStatus := ""
738 subsystem := ""
739 jobType := ""
740 if meta != nil {
741 if meta.qualifiedName != "" {
742 jobNameLabel = meta.qualifiedName
743 } else if meta.jobNumber != "" && meta.jobUser != "" && meta.jobName != "" {
744 jobNameLabel = fmt.Sprintf("%s/%s/%s", meta.jobNumber, meta.jobUser, meta.jobName)
745 } else if meta.jobName != "" {
746 jobNameLabel = meta.jobName
747 }
748 jobStatus = meta.jobStatus
749 subsystem = meta.subsystem
750 jobType = meta.jobType
751 }
752 labels := contexts.ActiveJobLabels{
753 Job_name: jobNameLabel,
754 Job_status: jobStatus,
755 Subsystem: subsystem,
756 Job_type: jobType,
757 }
758
759 contexts.ActiveJob.CPU.Set(c.State, labels, contexts.ActiveJobCPUValues{
760 Cpu: values.CPUPercentage,
761 })
762
763 contexts.ActiveJob.Resources.Set(c.State, labels, contexts.ActiveJobResourcesValues{
764 Temp_storage: values.TemporaryStorage,
765 })
766
767 contexts.ActiveJob.Time.Set(c.State, labels, contexts.ActiveJobTimeValues{
768 Cpu_time: values.ElapsedCPUTime,
769 Total_time: values.ElapsedTime,
770 })
771
772 contexts.ActiveJob.Activity.Set(c.State, labels, contexts.ActiveJobActivityValues{
773 Disk_io: values.ElapsedDiskIO,
774 Interactive_transactions: values.ElapsedInteractiveTransactions,
775 })
776
777 contexts.ActiveJob.Threads.Set(c.State, labels, contexts.ActiveJobThreadsValues{
778 Threads: values.ThreadCount,
779 })
780 }
781 }
782
783 func (c *Collector) exportNetworkInterfaceMetrics() {
784 for name, values := range c.mx.networkInterfaces {
785 meta := c.networkInterfaces[name]
786 interfaceType := ""
787 connectionType := ""
788 internetAddr := ""
789 networkAddr := ""
790 subnetMask := ""
791 if meta != nil {
792 interfaceType = meta.interfaceType
793 connectionType = meta.connectionType
794 internetAddr = meta.internetAddress
795 networkAddr = meta.networkAddress
796 subnetMask = meta.subnetMask
797 }
798 labels := contexts.NetworkInterfaceLabels{
799 Interface: name,
800 Interface_type: interfaceType,
801 Connection_type: connectionType,
802 Internet_address: internetAddr,
803 Network_address: networkAddr,
804 Subnet_mask: subnetMask,
805 }
806
807 contexts.NetworkInterface.Status.Set(c.State, labels, contexts.NetworkInterfaceStatusValues{
808 Active: values.InterfaceStatus,
809 })
810
811 contexts.NetworkInterface.MTU.Set(c.State, labels, contexts.NetworkInterfaceMTUValues{
812 Mtu: values.MTU,
813 })
814 }
815 }
816
817 func (c *Collector) exportHTTPServerMetrics() {
818 for key, values := range c.mx.httpServers {
819 meta := c.httpServers[key]
820 server := ""
821 function := ""
822 if meta != nil {
823 server = meta.serverName
824 function = meta.httpFunction
825 }
826 labels := contexts.HTTPServerLabels{
827 Server: server,
828 Function: function,
829 }
830 contexts.HTTPServer.Connections.Set(c.State, labels, contexts.HTTPServerConnectionsValues{
831 Normal: values.NormalConnections,
832 Ssl: values.SSLConnections,
833 })
834 contexts.HTTPServer.Threads.Set(c.State, labels, contexts.HTTPServerThreadsValues{
835 Active: values.ActiveThreads,
836 Idle: values.IdleThreads,
837 })
838 contexts.HTTPServer.Requests.Set(c.State, labels, contexts.HTTPServerRequestsValues{
839 Requests: values.TotalRequests,
840 Responses: values.TotalResponses,
841 Rejected: values.TotalRequestsRejected,
842 })
843 contexts.HTTPServer.Bytes.Set(c.State, labels, contexts.HTTPServerBytesValues{
844 Received: values.BytesReceived,
845 Sent: values.BytesSent,
846 })
847 }
848 }
849
850 func (c *Collector) exportPlanCacheMetrics() {
851 for key, values := range c.mx.planCache {
852 meta := c.planCache[key]
853 metricLabel := key
854 if meta != nil && meta.heading != "" {
855 metricLabel = meta.heading
856 }
857 labels := contexts.PlanCacheLabels{Metric: metricLabel}
858 if interval := c.slowPathIntervalSeconds(); interval > 0 {
859 contexts.PlanCache.Summary.SetUpdateEvery(c.State, labels, interval)
860 }
861 contexts.PlanCache.Summary.Set(c.State, labels, contexts.PlanCacheSummaryValues{
862 Value: values.Value,
863 })
864 }
865 }
866
867 func (c *Collector) logUnknownQueryLatency(path, name string) {
868 c.logOnce("unknown_query_latency_"+path+"_"+name, "%s path query latency not mapped to chart dimension: %s", path, name)
869 }
870
871 func (c *Collector) splitLatencyCounters(counters map[string]int64) (
872 contexts.ObservabilityQueryLatencyFastValues,
873 contexts.ObservabilityQueryLatencySlowValues,
874 contexts.ObservabilityQueryLatencyBatchValues,
875 bool, bool, bool,
876 ) {
877 var (
878 fast contexts.ObservabilityQueryLatencyFastValues
879 slow contexts.ObservabilityQueryLatencySlowValues
880 batch contexts.ObservabilityQueryLatencyBatchValues
881 )
882
883 if len(counters) == 0 {
884 return fast, slow, batch, false, false, false
885 }
886
887 fastDirect := map[string]*int64{
888 "count_disks": &fast.Count_disks,
889 "count_http_servers": &fast.Count_http_servers,
890 "count_network_interfaces": &fast.Count_network_interfaces,
891 "detect_ibmi_version_primary": &fast.Detect_ibmi_version_primary,
892 "detect_ibmi_version_fallback": &fast.Detect_ibmi_version_fallback,
893 "disk_instances": &fast.Disk_instances,
894 "disk_instances_enhanced": &fast.Disk_instances_enhanced,
895 "disk_status": &fast.Disk_status,
896 "http_server_info": &fast.Http_server_info,
897 "job_info": &fast.Job_info,
898 "memory_pools": &fast.Memory_pools,
899 "network_connections": &fast.Network_connections,
900 "network_interfaces": &fast.Network_interfaces,
901 "serial_number": &fast.Serial_number,
902 "system_activity": &fast.System_activity,
903 "system_model": &fast.System_model,
904 "system_status": &fast.System_status,
905 "system_name_metric": &fast.System_name,
906 "temp_storage_named": &fast.Temp_storage_named,
907 "temp_storage_total": &fast.Temp_storage_total,
908 "technology_refresh_level": &fast.Technology_refresh_level,
909 }
910
911 slowDirect := map[string]*int64{
912 "analyze_plan_cache": &slow.Analyze_plan_cache,
913 "count_subsystems": &slow.Count_subsystems,
914 "subsystems": &slow.Subsystems,
915 "plan_cache_summary": &slow.Plan_cache_summary,
916 }
917
918 var (
919 fastSet bool
920 slowSet bool
921 batchSet bool
922 )
923
924 for name, total := range counters {
925 if total <= 0 {
926 continue
927 }
928
929 if target, ok := fastDirect[name]; ok {
930 *target += total
931 fastSet = true
932 continue
933 }
934
935 if target, ok := slowDirect[name]; ok {
936 *target += total
937 slowSet = true
938 continue
939 }
940
941 switch {
942 case name == queryNameMessageQueueTotals:
943 batch.Message_queue_totals += total
944 batchSet = true
945 case name == queryNameJobQueueTotals:
946 batch.Job_queue_totals += total
947 batchSet = true
948 case name == queryNameOutputQueueTotals:
949 batch.Output_queue_totals += total
950 batchSet = true
951 case strings.HasPrefix(name, "message_queue_"):
952 slow.Message_queue_aggregates += total
953 slowSet = true
954 case strings.HasPrefix(name, "job_queue_"):
955 slow.Job_queues += total
956 slowSet = true
957 case strings.HasPrefix(name, "output_queue_"):
958 slow.Output_queue_info += total
959 slowSet = true
960 case strings.HasPrefix(name, "active_job_"):
961 fast.Active_job += total
962 fastSet = true
963 default:
964 c.logOnce("unknown_query_latency_"+name, "query latency not mapped to chart dimension: %s", name)
965 }
966 }
967
968 return fast, slow, batch, fastSet, slowSet, batchSet
969 }
970
971 func (c *Collector) exportQueryLatencyMetrics() {
972 fastValues, slowFallback, batchFallback, fastHasData, slowFallbackHasData, batchFallbackHasData := c.splitLatencyCounters(c.fastQueryLatencyCounters)
973
974 if fastHasData {
975 labels := contexts.EmptyLabels{}
976 if interval := c.fastPathIntervalSeconds(); interval > 0 {
977 contexts.Observability.QueryLatencyFast.SetUpdateEvery(c.State, labels, interval)
978 }
979 contexts.Observability.QueryLatencyFast.Set(c.State, labels, fastValues)
980 }
981
982 var slowSet bool
983 if c.slowPathActive() {
984 if counters, _ := c.slow.cache.getLatencies(); len(counters) > 0 {
985 _, slowValues, _, _, slowHasData, _ := c.splitLatencyCounters(counters)
986 if slowHasData {
987 labels := contexts.EmptyLabels{}
988 if interval := c.slowPathIntervalSeconds(); interval > 0 {
989 contexts.Observability.QueryLatencySlow.SetUpdateEvery(c.State, labels, interval)
990 }
991 contexts.Observability.QueryLatencySlow.Set(c.State, labels, slowValues)
992 slowSet = true
993 }
994 }
995 }
996 if !slowSet && slowFallbackHasData {
997 labels := contexts.EmptyLabels{}
998 if interval := c.fastPathIntervalSeconds(); interval > 0 {
999 contexts.Observability.QueryLatencySlow.SetUpdateEvery(c.State, labels, interval)
1000 }
1001 contexts.Observability.QueryLatencySlow.Set(c.State, labels, slowFallback)
1002 } else if !slowSet && c.slowPathActive() {
1003 labels := contexts.EmptyLabels{}
1004 if interval := c.slowPathIntervalSeconds(); interval > 0 {
1005 contexts.Observability.QueryLatencySlow.SetUpdateEvery(c.State, labels, interval)
1006 }
1007 contexts.Observability.QueryLatencySlow.Set(c.State, labels, contexts.ObservabilityQueryLatencySlowValues{})
1008 }
1009
1010 var batchSet bool
1011 if c.batchPathActive() {
1012 if counters, _ := c.batch.cache.getLatencies(); len(counters) > 0 {
1013 _, _, batchValues, _, _, batchHasData := c.splitLatencyCounters(counters)
1014 if batchHasData {
1015 c.batchLatencyValues = batchValues
1016 c.batchLatencyValid = true
1017 labels := contexts.EmptyLabels{}
1018 if interval := c.batchPathIntervalSeconds(); interval > 0 {
1019 contexts.Observability.QueryLatencyBatch.SetUpdateEvery(c.State, labels, interval)
1020 }
1021 contexts.Observability.QueryLatencyBatch.Set(c.State, labels, c.batchLatencyValues)
1022 batchSet = true
1023 }
1024 }
1025 }
1026 if !batchSet && c.batchLatencyValid {
1027 labels := contexts.EmptyLabels{}
1028 if interval := c.batchPathIntervalSeconds(); interval > 0 {
1029 contexts.Observability.QueryLatencyBatch.SetUpdateEvery(c.State, labels, interval)
1030 }
1031 contexts.Observability.QueryLatencyBatch.Set(c.State, labels, c.batchLatencyValues)
1032 batchSet = true
1033 }
1034 if !batchSet && batchFallbackHasData {
1035 labels := contexts.EmptyLabels{}
1036 if interval := c.fastPathIntervalSeconds(); interval > 0 {
1037 contexts.Observability.QueryLatencyBatch.SetUpdateEvery(c.State, labels, interval)
1038 }
1039 contexts.Observability.QueryLatencyBatch.Set(c.State, labels, batchFallback)
1040 } else if c.batchPathActive() && !batchSet {
1041 labels := contexts.EmptyLabels{}
1042 if interval := c.batchPathIntervalSeconds(); interval > 0 {
1043 contexts.Observability.QueryLatencyBatch.SetUpdateEvery(c.State, labels, interval)
1044 }
1045 contexts.Observability.QueryLatencyBatch.Set(c.State, labels, contexts.ObservabilityQueryLatencyBatchValues{})
1046 }
1047 }
1048
1049 func (c *Collector) fastPathIntervalSeconds() int {
1050 if c == nil {
1051 return 0
1052 }
1053 if c.Collector.Config.UpdateEvery > 0 {
1054 return c.Collector.Config.UpdateEvery
1055 }
1056 if c.Config.UpdateEvery > 0 {
1057 return c.Config.UpdateEvery
1058 }
1059 return 1
1060 }
1061
1062 func (c *Collector) slowPathIntervalSeconds() int {
1063 if c == nil {
1064 return 0
1065 }
1066 if !c.slowPathActive() {
1067 return 0
1068 }
1069 interval := int(c.slow.config.interval / time.Second)
1070 if interval < 1 {
1071 interval = max(c.fastPathIntervalSeconds(), 1)
1072 }
1073 return interval
1074 }
1075
1076 func (c *Collector) exportSystemActivityMetrics() {
1077 if c.mx.systemActivity.AverageCPURate == 0 && c.mx.systemActivity.AverageCPUUtilization == 0 {
1078 return
1079 }
1080 labels := contexts.EmptyLabels{}
1081 contexts.System.SystemActivityCPURate.Set(c.State, labels, contexts.SystemSystemActivityCPURateValues{
1082 Average: c.mx.systemActivity.AverageCPURate,
1083 })
1084 contexts.System.SystemActivityCPUUtilization.Set(c.State, labels, contexts.SystemSystemActivityCPUUtilizationValues{
1085 Average: c.mx.systemActivity.AverageCPUUtilization,
1086 Minimum: c.mx.systemActivity.MinimumCPUUtilization,
1087 Maximum: c.mx.systemActivity.MaximumCPUUtilization,
1088 })
1089 }
1090
1091 func (c *Collector) verifyConfig() error {
1092 if strings.TrimSpace(c.DSN) == "" {
1093 return errors.New("dsn is required but not set")
1094 }
1095 return nil
1096 }
1097
1098 func (c *Collector) Cleanup(ctx context.Context) {
1099 c.stopBatchPath()
1100 c.stopSlowPath()
1101 if c.client != nil {
1102 if err := c.client.Close(); err != nil {
1103 c.Errorf("cleanup: error closing database: %v", err)
1104 }
1105 }
1106 c.Collector.Cleanup(ctx)
1107 }
1108
1109 // EnableCaptureArtifacts allows the collector to emit structured capture artifacts when requested.
1110 func (c *Collector) EnableCaptureArtifacts(dir string) {
1111 ctx, err := newDumpContext(dir, &c.Config)
1112 if err != nil {
1113 c.Errorf("failed to initialise dump context: %v", err)
1114 return
1115 }
1116 c.dump = ctx
1117 c.Infof("dump data enabled, writing artifacts to %s", dir)
1118 }
1119
1120 func (c *Collector) buildDSNIfNeeded(ctx context.Context) error {
1121 if strings.TrimSpace(c.DSN) != "" {
1122 return nil
1123 }
1124
1125 if strings.TrimSpace(c.Hostname) == "" || strings.TrimSpace(c.Username) == "" || c.Password == "" {
1126 return fmt.Errorf("dsn required but not set, and insufficient connection parameters provided")
1127 }
1128
1129 cfg := &dbdriver.ConnectionConfig{
1130 Hostname: c.Hostname,
1131 Port: c.Port,
1132 Username: c.Username,
1133 Password: c.Password,
1134 Database: c.Database,
1135 SystemType: "AS400",
1136 ODBCDriver: c.ODBCDriver,
1137 UseSSL: c.UseSSL,
1138 }
1139
1140 c.DSN = dbdriver.BuildODBCDSN(cfg)
1141 return nil
1142 }