| 1 | package mq |
| 2 | |
| 3 | import ( |
| 4 | "github.com/netdata/netdata/go/plugins/plugin/ibm.d/modules/mq/contexts" |
| 5 | "github.com/netdata/netdata/go/plugins/plugin/ibm.d/protocols/pcf" |
| 6 | ) |
| 7 | |
| 8 | // collectStatistics collects advanced metrics from SYSTEM.ADMIN.STATISTICS.QUEUE |
| 9 | // This provides additional metrics like min/max depth, average queue time, and operation counts |
| 10 | // that are not available through regular PCF commands. |
| 11 | func (c *Collector) collectStatistics() error { |
| 12 | if !c.Config.CollectStatisticsQueue { |
| 13 | // Statistics collection is disabled |
| 14 | return nil |
| 15 | } |
| 16 | |
| 17 | c.Debugf("Collecting statistics from SYSTEM.ADMIN.STATISTICS.QUEUE") |
| 18 | |
| 19 | // Get statistics messages from the statistics queue |
| 20 | result, err := c.client.GetStatisticsQueue() |
| 21 | if err != nil { |
| 22 | c.Warningf("Failed to collect statistics from SYSTEM.ADMIN.STATISTICS.QUEUE: %v", err) |
| 23 | return nil // Don't fail the entire collection for statistics queue issues |
| 24 | } |
| 25 | |
| 26 | c.Debugf("Retrieved %d statistics messages", len(result.Messages)) |
| 27 | |
| 28 | // Process statistics messages |
| 29 | for _, msg := range result.Messages { |
| 30 | switch msg.Type { |
| 31 | case pcf.StatisticsTypeQueue: |
| 32 | if err := c.collectQueueStatistics(msg.QueueStats); err != nil { |
| 33 | c.Warningf("Failed to process queue statistics: %v", err) |
| 34 | } |
| 35 | case pcf.StatisticsTypeChannel: |
| 36 | if err := c.collectChannelStatistics(msg.ChannelStats); err != nil { |
| 37 | c.Warningf("Failed to process channel statistics: %v", err) |
| 38 | } |
| 39 | case pcf.StatisticsTypeMQI: |
| 40 | if err := c.collectMQIStatistics(msg.MQIStats); err != nil { |
| 41 | c.Warningf("Failed to process MQI statistics: %v", err) |
| 42 | } |
| 43 | default: |
| 44 | c.Debugf("Unknown statistics message type: %d", msg.Type) |
| 45 | } |
| 46 | } |
| 47 | |
| 48 | return nil |
| 49 | } |
| 50 | |
| 51 | // collectQueueStatistics processes queue statistics messages |
| 52 | func (c *Collector) collectQueueStatistics(queueStats []pcf.QueueStatistics) error { |
| 53 | for _, stat := range queueStats { |
| 54 | queueName := stat.Name |
| 55 | queueType := "local" // TODO: Convert stat.Type to string |
| 56 | |
| 57 | labels := contexts.QueueStatisticsLabels{ |
| 58 | Queue: queueName, |
| 59 | Type: queueType, |
| 60 | } |
| 61 | |
| 62 | // Min/Max depth metrics |
| 63 | if stat.MinDepth.IsCollected() && stat.MaxDepth.IsCollected() { |
| 64 | contexts.QueueStatistics.DepthMinMax.Set(c.State, labels, contexts.QueueStatisticsDepthMinMaxValues{ |
| 65 | Min_depth: stat.MinDepth.Int64(), |
| 66 | Max_depth: stat.MaxDepth.Int64(), |
| 67 | }) |
| 68 | contexts.QueueStatistics.DepthMinMax.SetUpdateEvery(c.State, labels, c.GetEffectiveStatisticsInterval()) |
| 69 | } |
| 70 | |
| 71 | // Average queue time metrics (split by persistence) |
| 72 | if stat.AvgQTimeNonPersistent.IsCollected() && stat.AvgQTimePersistent.IsCollected() { |
| 73 | contexts.QueueStatistics.AvgQueueTime.Set(c.State, labels, contexts.QueueStatisticsAvgQueueTimeValues{ |
| 74 | Non_persistent: stat.AvgQTimeNonPersistent.Int64(), |
| 75 | Persistent: stat.AvgQTimePersistent.Int64(), |
| 76 | }) |
| 77 | contexts.QueueStatistics.AvgQueueTime.SetUpdateEvery(c.State, labels, c.GetEffectiveStatisticsInterval()) |
| 78 | } |
| 79 | |
| 80 | // Queue time indicators (short/long period) |
| 81 | if stat.QTimeShort.IsCollected() && stat.QTimeLong.IsCollected() { |
| 82 | contexts.QueueStatistics.QueueTimeIndicators.Set(c.State, labels, contexts.QueueStatisticsQueueTimeIndicatorsValues{ |
| 83 | Short_period: stat.QTimeShort.Int64(), |
| 84 | Long_period: stat.QTimeLong.Int64(), |
| 85 | }) |
| 86 | contexts.QueueStatistics.QueueTimeIndicators.SetUpdateEvery(c.State, labels, c.GetEffectiveStatisticsInterval()) |
| 87 | } |
| 88 | |
| 89 | // Operation counters (incremental - rates per second) |
| 90 | contexts.QueueStatistics.Operations.Set(c.State, labels, contexts.QueueStatisticsOperationsValues{ |
| 91 | Puts_non_persistent: getValue(stat.PutsNonPersistent), |
| 92 | Puts_persistent: getValue(stat.PutsPersistent), |
| 93 | Gets_non_persistent: getValue(stat.GetsNonPersistent), |
| 94 | Gets_persistent: getValue(stat.GetsPersistent), |
| 95 | Put1s: getValue(stat.Put1Count), |
| 96 | Browses: getValue(stat.BrowseCount), |
| 97 | }) |
| 98 | contexts.QueueStatistics.Operations.SetUpdateEvery(c.State, labels, c.GetEffectiveStatisticsInterval()) |
| 99 | |
| 100 | // Byte counters (incremental - rates per second) |
| 101 | contexts.QueueStatistics.Bytes.Set(c.State, labels, contexts.QueueStatisticsBytesValues{ |
| 102 | Put_bytes_non_persistent: getValue(stat.PutBytesNonPersistent), |
| 103 | Put_bytes_persistent: getValue(stat.PutBytesPersistent), |
| 104 | Get_bytes_non_persistent: getValue(stat.GetBytesNonPersistent), |
| 105 | Get_bytes_persistent: getValue(stat.GetBytesPersistent), |
| 106 | Browse_bytes: getValue(stat.BrowseBytes), |
| 107 | }) |
| 108 | contexts.QueueStatistics.Bytes.SetUpdateEvery(c.State, labels, c.GetEffectiveStatisticsInterval()) |
| 109 | |
| 110 | // Failure counters (incremental - rates per second) |
| 111 | contexts.QueueStatistics.Failures.Set(c.State, labels, contexts.QueueStatisticsFailuresValues{ |
| 112 | Puts_failed: getValue(stat.PutsFailed), |
| 113 | Put1s_failed: getValue(stat.Put1sFailed), |
| 114 | Gets_failed: getValue(stat.GetsFailed), |
| 115 | Browses_failed: getValue(stat.BrowsesFailed), |
| 116 | }) |
| 117 | contexts.QueueStatistics.Failures.SetUpdateEvery(c.State, labels, c.GetEffectiveStatisticsInterval()) |
| 118 | |
| 119 | // Message lifecycle counters (incremental - rates per second) |
| 120 | contexts.QueueStatistics.MessageLifecycle.Set(c.State, labels, contexts.QueueStatisticsMessageLifecycleValues{ |
| 121 | Expired: getValue(stat.MsgsExpired), |
| 122 | Purged: getValue(stat.MsgsPurged), |
| 123 | Not_queued: getValue(stat.MsgsNotQueued), |
| 124 | }) |
| 125 | contexts.QueueStatistics.MessageLifecycle.SetUpdateEvery(c.State, labels, c.GetEffectiveStatisticsInterval()) |
| 126 | |
| 127 | c.Debugf("Collected statistics for queue '%s' (type: %s)", queueName, queueType) |
| 128 | } |
| 129 | |
| 130 | return nil |
| 131 | } |
| 132 | |
| 133 | // collectChannelStatistics processes channel statistics messages |
| 134 | func (c *Collector) collectChannelStatistics(channelStats []pcf.ChannelStatistics) error { |
| 135 | for _, stat := range channelStats { |
| 136 | channelName := stat.Name |
| 137 | channelType := "svrconn" // TODO: Convert stat.Type to string |
| 138 | |
| 139 | labels := contexts.ChannelStatisticsLabels{ |
| 140 | Channel: channelName, |
| 141 | Type: channelType, |
| 142 | } |
| 143 | |
| 144 | // Message metrics (incremental - rates per second) |
| 145 | if stat.Messages.IsCollected() { |
| 146 | contexts.ChannelStatistics.Messages.Set(c.State, labels, contexts.ChannelStatisticsMessagesValues{ |
| 147 | Messages: stat.Messages.Int64(), |
| 148 | }) |
| 149 | contexts.ChannelStatistics.Messages.SetUpdateEvery(c.State, labels, c.GetEffectiveStatisticsInterval()) |
| 150 | } |
| 151 | |
| 152 | // Byte metrics (incremental - rates per second) |
| 153 | if stat.Bytes.IsCollected() { |
| 154 | contexts.ChannelStatistics.Bytes.Set(c.State, labels, contexts.ChannelStatisticsBytesValues{ |
| 155 | Bytes: stat.Bytes.Int64(), |
| 156 | }) |
| 157 | contexts.ChannelStatistics.Bytes.SetUpdateEvery(c.State, labels, c.GetEffectiveStatisticsInterval()) |
| 158 | } |
| 159 | |
| 160 | // Batch metrics (incremental - rates per second) |
| 161 | contexts.ChannelStatistics.Batches.Set(c.State, labels, contexts.ChannelStatisticsBatchesValues{ |
| 162 | Full_batches: getValue(stat.FullBatches), |
| 163 | Incomplete_batches: getValue(stat.IncompleteBatches), |
| 164 | }) |
| 165 | contexts.ChannelStatistics.Batches.SetUpdateEvery(c.State, labels, c.GetEffectiveStatisticsInterval()) |
| 166 | |
| 167 | // Average batch size (absolute value) |
| 168 | if stat.AvgBatchSize.IsCollected() { |
| 169 | contexts.ChannelStatistics.BatchSize.Set(c.State, labels, contexts.ChannelStatisticsBatchSizeValues{ |
| 170 | Avg_batch_size: stat.AvgBatchSize.Int64(), |
| 171 | }) |
| 172 | contexts.ChannelStatistics.BatchSize.SetUpdateEvery(c.State, labels, c.GetEffectiveStatisticsInterval()) |
| 173 | } |
| 174 | |
| 175 | // Put retry metrics (incremental - rates per second) |
| 176 | if stat.PutRetries.IsCollected() { |
| 177 | contexts.ChannelStatistics.PutRetries.Set(c.State, labels, contexts.ChannelStatisticsPutRetriesValues{ |
| 178 | Put_retries: stat.PutRetries.Int64(), |
| 179 | }) |
| 180 | contexts.ChannelStatistics.PutRetries.SetUpdateEvery(c.State, labels, c.GetEffectiveStatisticsInterval()) |
| 181 | } |
| 182 | |
| 183 | c.Debugf("Collected statistics for channel '%s' (type: %s)", channelName, channelType) |
| 184 | } |
| 185 | |
| 186 | return nil |
| 187 | } |
| 188 | |
| 189 | // collectMQIStatistics processes MQI statistics messages |
| 190 | func (c *Collector) collectMQIStatistics(mqiStats []pcf.MQIStatistics) error { |
| 191 | for _, stat := range mqiStats { |
| 192 | // MQI stats are at queue manager level |
| 193 | labels := contexts.MQIStatisticsLabels{ |
| 194 | Queue_manager: stat.Name, |
| 195 | } |
| 196 | |
| 197 | // MQOPEN operations (incremental - rates per second) |
| 198 | if stat.Opens.IsCollected() || stat.OpensFailed.IsCollected() { |
| 199 | contexts.MQIStatistics.Opens.Set(c.State, labels, contexts.MQIStatisticsOpensValues{ |
| 200 | Opens_total: getValue(stat.Opens), |
| 201 | Opens_failed: getValue(stat.OpensFailed), |
| 202 | }) |
| 203 | contexts.MQIStatistics.Opens.SetUpdateEvery(c.State, labels, c.GetEffectiveStatisticsInterval()) |
| 204 | } |
| 205 | |
| 206 | // MQCLOSE operations (incremental - rates per second) |
| 207 | if stat.Closes.IsCollected() || stat.ClosesFailed.IsCollected() { |
| 208 | contexts.MQIStatistics.Closes.Set(c.State, labels, contexts.MQIStatisticsClosesValues{ |
| 209 | Closes_total: getValue(stat.Closes), |
| 210 | Closes_failed: getValue(stat.ClosesFailed), |
| 211 | }) |
| 212 | contexts.MQIStatistics.Closes.SetUpdateEvery(c.State, labels, c.GetEffectiveStatisticsInterval()) |
| 213 | } |
| 214 | |
| 215 | // MQINQ operations (incremental - rates per second) |
| 216 | if stat.Inqs.IsCollected() || stat.InqsFailed.IsCollected() { |
| 217 | contexts.MQIStatistics.Inqs.Set(c.State, labels, contexts.MQIStatisticsInqsValues{ |
| 218 | Inqs_total: getValue(stat.Inqs), |
| 219 | Inqs_failed: getValue(stat.InqsFailed), |
| 220 | }) |
| 221 | contexts.MQIStatistics.Inqs.SetUpdateEvery(c.State, labels, c.GetEffectiveStatisticsInterval()) |
| 222 | } |
| 223 | |
| 224 | // MQSET operations (incremental - rates per second) |
| 225 | if stat.Sets.IsCollected() || stat.SetsFailed.IsCollected() { |
| 226 | contexts.MQIStatistics.Sets.Set(c.State, labels, contexts.MQIStatisticsSetsValues{ |
| 227 | Sets_total: getValue(stat.Sets), |
| 228 | Sets_failed: getValue(stat.SetsFailed), |
| 229 | }) |
| 230 | contexts.MQIStatistics.Sets.SetUpdateEvery(c.State, labels, c.GetEffectiveStatisticsInterval()) |
| 231 | } |
| 232 | |
| 233 | c.Debugf("Collected MQI statistics for queue manager '%s'", stat.Name) |
| 234 | } |
| 235 | |
| 236 | return nil |
| 237 | } |
| 238 | |
| 239 | // getValue returns the int64 value of an AttributeValue, or 0 if not collected |
| 240 | func getValue(attr pcf.AttributeValue) int64 { |
| 241 | if attr.IsCollected() { |
| 242 | return attr.Int64() |
| 243 | } |
| 244 | return 0 |
| 245 | } |