| 1 | package mq |
| 2 | |
| 3 | import ( |
| 4 | "fmt" |
| 5 | |
| 6 | "github.com/netdata/netdata/go/plugins/plugin/ibm.d/modules/mq/contexts" |
| 7 | "github.com/netdata/netdata/go/plugins/plugin/ibm.d/protocols/pcf" |
| 8 | ) |
| 9 | |
| 10 | func (c *Collector) collectChannelMetrics() error { |
| 11 | c.Debugf("Collecting channels with selector '%s', config: %v, system: %v", |
| 12 | c.Config.ChannelSelector, c.Config.CollectChannelConfig, c.Config.CollectSystemChannels) |
| 13 | |
| 14 | // Use new GetChannels with transparency |
| 15 | result, err := c.client.GetChannels( |
| 16 | c.Config.CollectChannelConfig, // collectConfig |
| 17 | true, // collectMetrics (always) |
| 18 | c.Config.MaxChannels, // maxChannels (0 = no limit) |
| 19 | c.Config.ChannelSelector, // selector pattern |
| 20 | c.Config.CollectSystemChannels, // collectSystem |
| 21 | ) |
| 22 | if err != nil { |
| 23 | return fmt.Errorf("failed to collect channel metrics: %w", err) |
| 24 | } |
| 25 | |
| 26 | // Check discovery success |
| 27 | if !result.Stats.Discovery.Success { |
| 28 | c.Errorf("Channel discovery failed completely") |
| 29 | return fmt.Errorf("channel discovery failed") |
| 30 | } |
| 31 | |
| 32 | // Map transparency counters to user-facing semantics |
| 33 | monitored := int64(0) |
| 34 | if result.Stats.Metrics != nil { |
| 35 | monitored = result.Stats.Metrics.OkItems |
| 36 | } |
| 37 | |
| 38 | failed := result.Stats.Discovery.UnparsedItems |
| 39 | if result.Stats.Metrics != nil { |
| 40 | failed += result.Stats.Metrics.FailedItems |
| 41 | } |
| 42 | // Note: Config failures are not counted as they're optional |
| 43 | |
| 44 | // Update overview metrics with correct semantics |
| 45 | c.setChannelOverviewMetrics( |
| 46 | monitored, // monitored (successfully enriched) |
| 47 | result.Stats.Discovery.ExcludedItems, // excluded (filtered by user) |
| 48 | result.Stats.Discovery.InvisibleItems, // invisible (discovery errors) |
| 49 | failed, // failed (unparsed + enrichment failures) |
| 50 | ) |
| 51 | |
| 52 | // Log collection summary |
| 53 | c.Debugf("Channel collection complete - discovered:%d visible:%d included:%d collected:%d failed:%d", |
| 54 | result.Stats.Discovery.AvailableItems, |
| 55 | result.Stats.Discovery.AvailableItems-result.Stats.Discovery.InvisibleItems, |
| 56 | result.Stats.Discovery.IncludedItems, |
| 57 | len(result.Channels), |
| 58 | failed) |
| 59 | |
| 60 | // Process collected channel metrics |
| 61 | for _, channel := range result.Channels { |
| 62 | labels := contexts.ChannelLabels{ |
| 63 | Channel: channel.Name, |
| 64 | Type: channel.Type.String(), // Add the channel type |
| 65 | } |
| 66 | |
| 67 | // Set channel status - convert enum to individual metrics |
| 68 | statusValues := contexts.ChannelStatusValues{} |
| 69 | |
| 70 | // Set the active status to 1 |
| 71 | switch channel.Status { |
| 72 | case pcf.ChannelStatusInactive: |
| 73 | statusValues.Inactive = 1 |
| 74 | case pcf.ChannelStatusBinding: |
| 75 | statusValues.Binding = 1 |
| 76 | case pcf.ChannelStatusStarting: |
| 77 | statusValues.Starting = 1 |
| 78 | case pcf.ChannelStatusRunning: |
| 79 | statusValues.Running = 1 |
| 80 | case pcf.ChannelStatusStopping: |
| 81 | statusValues.Stopping = 1 |
| 82 | case pcf.ChannelStatusRetrying: |
| 83 | statusValues.Retrying = 1 |
| 84 | case pcf.ChannelStatusStopped: |
| 85 | statusValues.Stopped = 1 |
| 86 | case pcf.ChannelStatusRequesting: |
| 87 | statusValues.Requesting = 1 |
| 88 | case pcf.ChannelStatusPaused: |
| 89 | statusValues.Paused = 1 |
| 90 | case pcf.ChannelStatusDisconnected: |
| 91 | statusValues.Disconnected = 1 |
| 92 | case pcf.ChannelStatusInitializing: |
| 93 | statusValues.Initializing = 1 |
| 94 | case pcf.ChannelStatusSwitching: |
| 95 | statusValues.Switching = 1 |
| 96 | } |
| 97 | |
| 98 | contexts.Channel.Status.Set(c.State, labels, statusValues) |
| 99 | |
| 100 | // Set channel messages (incremental) - only if available |
| 101 | if channel.Messages != nil { |
| 102 | contexts.Channel.Messages.Set(c.State, labels, contexts.ChannelMessagesValues{ |
| 103 | Messages: *channel.Messages, |
| 104 | }) |
| 105 | } |
| 106 | |
| 107 | // Set channel bytes (incremental) - only if available |
| 108 | if channel.Bytes != nil { |
| 109 | contexts.Channel.Bytes.Set(c.State, labels, contexts.ChannelBytesValues{ |
| 110 | Bytes: *channel.Bytes, |
| 111 | }) |
| 112 | } |
| 113 | |
| 114 | // Set channel batches (incremental) - only if available |
| 115 | if channel.Batches != nil { |
| 116 | contexts.Channel.Batches.Set(c.State, labels, contexts.ChannelBatchesValues{ |
| 117 | Batches: *channel.Batches, |
| 118 | }) |
| 119 | } |
| 120 | |
| 121 | // Configuration metrics - only set when configuration collection is enabled |
| 122 | if c.Config.CollectChannelConfig { |
| 123 | // Set batch size - only if available |
| 124 | if channel.BatchSize.IsCollected() { |
| 125 | contexts.Channel.BatchSize.Set(c.State, labels, contexts.ChannelBatchSizeValues{ |
| 126 | Batch_size: channel.BatchSize.Int64(), |
| 127 | }) |
| 128 | } |
| 129 | |
| 130 | // Set batch interval - only if available |
| 131 | if channel.BatchInterval.IsCollected() { |
| 132 | contexts.Channel.BatchInterval.Set(c.State, labels, contexts.ChannelBatchIntervalValues{ |
| 133 | Batch_interval: channel.BatchInterval.Int64(), |
| 134 | }) |
| 135 | } |
| 136 | |
| 137 | // Set intervals - only if at least one is available and not -1 (which means disabled/not applicable) |
| 138 | intervalValues := contexts.ChannelIntervalsValues{} |
| 139 | hasAnyInterval := false |
| 140 | |
| 141 | if channel.DiscInterval.IsCollected() && channel.DiscInterval.Int64() != -1 { |
| 142 | intervalValues.Disc_interval = channel.DiscInterval.Int64() |
| 143 | hasAnyInterval = true |
| 144 | } |
| 145 | if channel.HbInterval.IsCollected() && channel.HbInterval.Int64() != -1 { |
| 146 | intervalValues.Hb_interval = channel.HbInterval.Int64() |
| 147 | hasAnyInterval = true |
| 148 | } |
| 149 | if channel.KeepAliveInterval.IsCollected() && channel.KeepAliveInterval.Int64() != -1 { |
| 150 | intervalValues.Keep_alive_interval = channel.KeepAliveInterval.Int64() |
| 151 | hasAnyInterval = true |
| 152 | } |
| 153 | |
| 154 | // Only send the metric if we have at least one valid (non -1) interval |
| 155 | if hasAnyInterval { |
| 156 | contexts.Channel.Intervals.Set(c.State, labels, intervalValues) |
| 157 | } |
| 158 | |
| 159 | // Set short retry count - only if available |
| 160 | if channel.ShortRetry.IsCollected() { |
| 161 | contexts.Channel.ShortRetryCount.Set(c.State, labels, contexts.ChannelShortRetryCountValues{ |
| 162 | Short_retry: channel.ShortRetry.Int64(), |
| 163 | }) |
| 164 | } |
| 165 | |
| 166 | // Set long retry interval - only if available |
| 167 | if channel.LongRetry.IsCollected() { |
| 168 | contexts.Channel.LongRetryInterval.Set(c.State, labels, contexts.ChannelLongRetryIntervalValues{ |
| 169 | Long_retry: channel.LongRetry.Int64(), |
| 170 | }) |
| 171 | } |
| 172 | |
| 173 | // Set max message length - only if available |
| 174 | if channel.MaxMsgLength.IsCollected() { |
| 175 | contexts.Channel.MaxMessageLength.Set(c.State, labels, contexts.ChannelMaxMessageLengthValues{ |
| 176 | Max_msg_length: channel.MaxMsgLength.Int64(), |
| 177 | }) |
| 178 | } |
| 179 | |
| 180 | // Set sharing conversations - only if available |
| 181 | if channel.SharingConversations.IsCollected() { |
| 182 | contexts.Channel.SharingConversations.Set(c.State, labels, contexts.ChannelSharingConversationsValues{ |
| 183 | Sharing_conversations: channel.SharingConversations.Int64(), |
| 184 | }) |
| 185 | } |
| 186 | |
| 187 | // Set network priority - only if available |
| 188 | if channel.NetworkPriority.IsCollected() { |
| 189 | contexts.Channel.NetworkPriority.Set(c.State, labels, contexts.ChannelNetworkPriorityValues{ |
| 190 | Network_priority: channel.NetworkPriority.Int64(), |
| 191 | }) |
| 192 | } |
| 193 | } |
| 194 | |
| 195 | // Extended status metrics - only send if collected and available for all channels |
| 196 | |
| 197 | // Buffer counts - only send if at least one is collected |
| 198 | if channel.BuffersSent.IsCollected() || channel.BuffersReceived.IsCollected() { |
| 199 | bufferValues := contexts.ChannelBufferCountsValues{} |
| 200 | hasAnyBuffer := false |
| 201 | |
| 202 | if channel.BuffersSent.IsCollected() { |
| 203 | bufferValues.Sent = channel.BuffersSent.Int64() |
| 204 | hasAnyBuffer = true |
| 205 | } |
| 206 | if channel.BuffersReceived.IsCollected() { |
| 207 | bufferValues.Received = channel.BuffersReceived.Int64() |
| 208 | hasAnyBuffer = true |
| 209 | } |
| 210 | |
| 211 | if hasAnyBuffer { |
| 212 | contexts.Channel.BufferCounts.Set(c.State, labels, bufferValues) |
| 213 | } |
| 214 | } |
| 215 | |
| 216 | // Current messages - only if collected |
| 217 | if channel.CurrentMessages.IsCollected() { |
| 218 | contexts.Channel.CurrentMessages.Set(c.State, labels, contexts.ChannelCurrentMessagesValues{ |
| 219 | Current: channel.CurrentMessages.Int64(), |
| 220 | }) |
| 221 | } |
| 222 | |
| 223 | // XMITQ time indicator - only if collected |
| 224 | if channel.XmitQueueTime.IsCollected() { |
| 225 | contexts.Channel.XmitQueueTime.Set(c.State, labels, contexts.ChannelXmitQueueTimeValues{ |
| 226 | Xmitq_time: channel.XmitQueueTime.Int64(), |
| 227 | }) |
| 228 | } |
| 229 | |
| 230 | // MCA status - only if collected |
| 231 | if channel.MCAStatus.IsCollected() { |
| 232 | contexts.Channel.MCAStatus.Set(c.State, labels, contexts.ChannelMCAStatusValues{ |
| 233 | Mca_status: channel.MCAStatus.Int64(), |
| 234 | }) |
| 235 | } |
| 236 | |
| 237 | // In-doubt status - only if collected |
| 238 | if channel.InDoubtStatus.IsCollected() { |
| 239 | contexts.Channel.InDoubtStatus.Set(c.State, labels, contexts.ChannelInDoubtStatusValues{ |
| 240 | Indoubt_status: channel.InDoubtStatus.Int64(), |
| 241 | }) |
| 242 | } |
| 243 | |
| 244 | // SSL key resets - only if collected |
| 245 | if channel.SSLKeyResets.IsCollected() { |
| 246 | contexts.Channel.SSLKeyResets.Set(c.State, labels, contexts.ChannelSSLKeyResetsValues{ |
| 247 | Ssl_key_resets: channel.SSLKeyResets.Int64(), |
| 248 | }) |
| 249 | } |
| 250 | |
| 251 | // NPM speed - only if collected |
| 252 | if channel.NPMSpeed.IsCollected() { |
| 253 | contexts.Channel.NPMSpeed.Set(c.State, labels, contexts.ChannelNPMSpeedValues{ |
| 254 | Npm_speed: channel.NPMSpeed.Int64(), |
| 255 | }) |
| 256 | } |
| 257 | |
| 258 | // Current sharing conversations - only if collected |
| 259 | if channel.CurrentSharingConvs.IsCollected() { |
| 260 | contexts.Channel.CurrentSharingConversations.Set(c.State, labels, contexts.ChannelCurrentSharingConversationsValues{ |
| 261 | Current_sharing: channel.CurrentSharingConvs.Int64(), |
| 262 | }) |
| 263 | } |
| 264 | } |
| 265 | |
| 266 | return nil |
| 267 | } |