| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | //go:build cgo && ibm_mq |
| 4 | |
| 5 | package pcf |
| 6 | |
| 7 | import ( |
| 8 | "fmt" |
| 9 | |
| 10 | "github.com/ibm-messaging/mq-golang/v5/mqmetric" |
| 11 | ) |
| 12 | |
| 13 | // ResourceMetric represents a single resource metric from IBM mqmetric |
| 14 | type ResourceMetric struct { |
| 15 | Class string |
| 16 | Type string |
| 17 | Element string |
| 18 | Instance string |
| 19 | Value interface{} |
| 20 | MetricName string |
| 21 | } |
| 22 | |
| 23 | // CollectResourceMetrics collects resource metrics using IBM mqmetric library |
| 24 | func (c *Client) CollectResourceMetrics() (map[string]ResourceMetric, error) { |
| 25 | if !c.metricsReady { |
| 26 | return nil, fmt.Errorf("resource monitoring not initialized") |
| 27 | } |
| 28 | |
| 29 | // Get all published metrics from IBM library |
| 30 | // This uses mqmetric's connection (Connection #2) |
| 31 | allMetrics := mqmetric.GetPublishedMetrics(mqmetric.GetConnectionKey()) |
| 32 | |
| 33 | result := make(map[string]ResourceMetric) |
| 34 | |
| 35 | // Process IBM's metric hierarchy |
| 36 | for classKey, class := range allMetrics.Classes { |
| 37 | for typeKey, mType := range class.Types { |
| 38 | for elemKey, element := range mType.Elements { |
| 39 | for instanceName, value := range element.Values { |
| 40 | key := fmt.Sprintf("%s.%s.%s.%s", classKey, typeKey, elemKey, instanceName) |
| 41 | result[key] = ResourceMetric{ |
| 42 | Class: class.Name, |
| 43 | Type: mType.Name, |
| 44 | Element: element.Description, |
| 45 | Instance: instanceName, |
| 46 | Value: value, |
| 47 | MetricName: element.MetricName, |
| 48 | } |
| 49 | } |
| 50 | } |
| 51 | } |
| 52 | } |
| 53 | |
| 54 | c.protocol.Debugf("collected %d resource metrics", len(result)) |
| 55 | return result, nil |
| 56 | } |
| 57 | |
| 58 | // DiscoverAndSubscribeResources discovers and subscribes to resource metrics |
| 59 | func (c *Client) DiscoverAndSubscribeResources(config ResourceDiscoveryConfig) error { |
| 60 | if !c.metricsReady { |
| 61 | return fmt.Errorf("resource monitoring not initialized") |
| 62 | } |
| 63 | |
| 64 | // Configure discovery |
| 65 | c.resourceConfig.MonitoredQueues.ObjectNames = config.QueueSelector |
| 66 | c.resourceConfig.MonitoredQueues.UseWildcard = true |
| 67 | |
| 68 | // Discover and subscribe using mqmetric |
| 69 | err := mqmetric.DiscoverAndSubscribe(*c.resourceConfig) |
| 70 | if err != nil { |
| 71 | return fmt.Errorf("failed to discover and subscribe to resources: %w", err) |
| 72 | } |
| 73 | |
| 74 | c.protocol.Debugf("resource discovery and subscription completed") |
| 75 | return nil |
| 76 | } |
| 77 | |
| 78 | // ResourceDiscoveryConfig configuration for resource discovery |
| 79 | type ResourceDiscoveryConfig struct { |
| 80 | QueueSelector string |
| 81 | EnableStats bool |
| 82 | } |
| 83 | |
| 84 | // ResourcePublicationsResult contains the result of resource monitoring queries |
| 85 | type ResourcePublicationsResult struct { |
| 86 | Stats CollectionStats |
| 87 | UserCPUPercent AttributeValue |
| 88 | SystemCPUPercent AttributeValue |
| 89 | AvailableMemory AttributeValue |
| 90 | UsedMemory AttributeValue |
| 91 | MemoryUsedMB AttributeValue |
| 92 | LogUsedBytes AttributeValue |
| 93 | LogMaxBytes AttributeValue |
| 94 | // Additional resource metrics can be added here as needed |
| 95 | } |
| 96 | |
| 97 | // IsResourceMonitoringSupported checks if resource monitoring is supported |
| 98 | func (c *Client) IsResourceMonitoringSupported() bool { |
| 99 | // Resource monitoring requires MQ v9+ and is available on distributed platforms |
| 100 | // For now, we'll assume it's supported if we're connected |
| 101 | // Real implementation would check the command level and platform |
| 102 | return c.connected && c.cachedCommandLevel >= 900 |
| 103 | } |
| 104 | |
| 105 | // EnableResourceMonitoring enables resource monitoring for the queue manager |
| 106 | func (c *Client) EnableResourceMonitoring() error { |
| 107 | if c.resourceStatus == ResourceStatusFailed { |
| 108 | return fmt.Errorf("resource monitoring permanently disabled") |
| 109 | } |
| 110 | |
| 111 | if c.resourceStatus == ResourceStatusEnabled { |
| 112 | // Already enabled |
| 113 | return nil |
| 114 | } |
| 115 | |
| 116 | // Initialize the metrics connection (Connection #2) |
| 117 | if !c.metricsReady { |
| 118 | connConfig := mqmetric.ConnectionConfig{ |
| 119 | ClientMode: true, |
| 120 | UserId: c.config.User, |
| 121 | Password: c.config.Password, |
| 122 | UsePublications: true, |
| 123 | WaitInterval: 30, |
| 124 | ConnName: fmt.Sprintf("%s(%d)", c.config.Host, c.config.Port), |
| 125 | Channel: c.config.Channel, |
| 126 | } |
| 127 | |
| 128 | err := mqmetric.InitConnection(c.config.QueueManager, "NETDATA.REPLY.METRICS", "", &connConfig) |
| 129 | if err != nil { |
| 130 | c.resourceStatus = ResourceStatusFailed |
| 131 | return fmt.Errorf("failed to initialize metrics connection: %w", err) |
| 132 | } |
| 133 | |
| 134 | c.metricsReady = true |
| 135 | } |
| 136 | |
| 137 | // Set up basic discovery configuration |
| 138 | c.resourceConfig = &mqmetric.DiscoverConfig{ |
| 139 | MetaPrefix: "$SYS/MQ/INFO", |
| 140 | } |
| 141 | |
| 142 | c.resourceStatus = ResourceStatusEnabled |
| 143 | c.protocol.Debugf("resource monitoring enabled") |
| 144 | return nil |
| 145 | } |
| 146 | |
| 147 | // GetResourcePublications retrieves resource monitoring data |
| 148 | func (c *Client) GetResourcePublications() (*ResourcePublicationsResult, error) { |
| 149 | if c.resourceStatus != ResourceStatusEnabled || !c.metricsReady { |
| 150 | return nil, fmt.Errorf("resource monitoring not enabled") |
| 151 | } |
| 152 | |
| 153 | // Get metrics from IBM mqmetric library |
| 154 | metrics, err := c.CollectResourceMetrics() |
| 155 | if err != nil { |
| 156 | return nil, fmt.Errorf("failed to collect resource metrics: %w", err) |
| 157 | } |
| 158 | |
| 159 | // Create result structure |
| 160 | result := &ResourcePublicationsResult{ |
| 161 | Stats: CollectionStats{ |
| 162 | Discovery: struct { |
| 163 | Success bool |
| 164 | AvailableItems int64 |
| 165 | InvisibleItems int64 |
| 166 | IncludedItems int64 |
| 167 | ExcludedItems int64 |
| 168 | UnparsedItems int64 |
| 169 | ErrorCounts map[int32]int |
| 170 | }{ |
| 171 | Success: true, |
| 172 | AvailableItems: int64(len(metrics)), |
| 173 | IncludedItems: int64(len(metrics)), |
| 174 | ErrorCounts: make(map[int32]int), |
| 175 | }, |
| 176 | }, |
| 177 | UserCPUPercent: NotCollected, |
| 178 | SystemCPUPercent: NotCollected, |
| 179 | AvailableMemory: NotCollected, |
| 180 | UsedMemory: NotCollected, |
| 181 | MemoryUsedMB: NotCollected, |
| 182 | LogUsedBytes: NotCollected, |
| 183 | LogMaxBytes: NotCollected, |
| 184 | } |
| 185 | |
| 186 | // Process specific metrics we care about |
| 187 | for key, metric := range metrics { |
| 188 | if value, ok := metric.Value.(float64); ok { |
| 189 | switch { |
| 190 | case metric.Element == "User CPU time percentage": |
| 191 | result.UserCPUPercent = AttributeValue(int64(value * 1000)) // Convert to 3 decimal precision |
| 192 | case metric.Element == "System CPU time percentage": |
| 193 | result.SystemCPUPercent = AttributeValue(int64(value * 1000)) |
| 194 | case metric.Element == "Available memory": |
| 195 | result.AvailableMemory = AttributeValue(int64(value)) |
| 196 | case metric.Element == "Used memory": |
| 197 | result.UsedMemory = AttributeValue(int64(value)) |
| 198 | case metric.Element == "RAM total bytes for queue manager": |
| 199 | result.MemoryUsedMB = AttributeValue(int64(value / (1024 * 1024))) // Convert bytes to MB |
| 200 | case metric.Element == "Log - bytes in use": |
| 201 | result.LogUsedBytes = AttributeValue(int64(value)) |
| 202 | case metric.Element == "Log - bytes max": |
| 203 | result.LogMaxBytes = AttributeValue(int64(value)) |
| 204 | } |
| 205 | } |
| 206 | c.protocol.Debugf("processed resource metric: %s = %v", key, metric.Value) |
| 207 | } |
| 208 | |
| 209 | return result, nil |
| 210 | } |
| 211 | |
| 212 | // InitializeResourceDiscovery initializes resource discovery with custom configuration |
| 213 | func (c *Client) InitializeResourceDiscovery(config ResourceDiscoveryConfig) error { |
| 214 | if !c.metricsReady { |
| 215 | return fmt.Errorf("resource monitoring connection not available") |
| 216 | } |
| 217 | |
| 218 | // Configure discovery settings |
| 219 | discoveryConfig := &mqmetric.DiscoverConfig{ |
| 220 | MetaPrefix: "$SYS/MQ/INFO", |
| 221 | MonitoredQueues: mqmetric.DiscoverObject{ |
| 222 | ObjectNames: config.QueueSelector, |
| 223 | UseWildcard: true, |
| 224 | SubscriptionSelector: "", |
| 225 | }, |
| 226 | } |
| 227 | |
| 228 | // Update connection config if needed |
| 229 | connConfig := mqmetric.ConnectionConfig{ |
| 230 | ClientMode: true, |
| 231 | UserId: c.config.User, |
| 232 | Password: c.config.Password, |
| 233 | UsePublications: true, |
| 234 | UseResetQStats: config.EnableStats, |
| 235 | WaitInterval: 30, |
| 236 | ConnName: fmt.Sprintf("%s(%d)", c.config.Host, c.config.Port), |
| 237 | Channel: c.config.Channel, |
| 238 | } |
| 239 | |
| 240 | // Re-initialize with new configuration |
| 241 | mqmetric.EndConnection() |
| 242 | err := mqmetric.InitConnection(c.config.QueueManager, "NETDATA.REPLY.METRICS", "", &connConfig) |
| 243 | if err != nil { |
| 244 | c.metricsReady = false |
| 245 | return fmt.Errorf("failed to reinitialize metrics connection: %w", err) |
| 246 | } |
| 247 | |
| 248 | // Perform discovery and subscription |
| 249 | err = mqmetric.DiscoverAndSubscribe(*discoveryConfig) |
| 250 | if err != nil { |
| 251 | return fmt.Errorf("failed to discover and subscribe: %w", err) |
| 252 | } |
| 253 | |
| 254 | c.resourceConfig = discoveryConfig |
| 255 | c.protocol.Debugf("resource discovery initialized with queue selector: %s", config.QueueSelector) |
| 256 | |
| 257 | return nil |
| 258 | } |