| 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 | "path/filepath" |
| 10 | "strings" |
| 11 | |
| 12 | "github.com/ibm-messaging/mq-golang/v5/ibmmq" |
| 13 | ) |
| 14 | |
| 15 | // GetChannelList returns a list of channels. |
| 16 | func (c *Client) GetChannelList() ([]string, error) { |
| 17 | c.protocol.Debugf("Getting channel list from queue manager '%s'", c.config.QueueManager) |
| 18 | |
| 19 | const pattern = "*" |
| 20 | params := []pcfParameter{ |
| 21 | newStringParameter(ibmmq.MQCACH_CHANNEL_NAME, pattern), |
| 22 | } |
| 23 | response, err := c.sendPCFCommand(ibmmq.MQCMD_INQUIRE_CHANNEL, params) |
| 24 | if err != nil { |
| 25 | c.protocol.Errorf("Failed to get channel list from queue manager '%s': %v", c.config.QueueManager, err) |
| 26 | return nil, err |
| 27 | } |
| 28 | |
| 29 | result := c.parseChannelListResponseFromParams(response) |
| 30 | |
| 31 | if result.InternalErrors > 0 { |
| 32 | c.protocol.Warningf("Encountered %d internal errors while parsing channel list from queue manager '%s'", |
| 33 | result.InternalErrors, c.config.QueueManager) |
| 34 | } |
| 35 | |
| 36 | for errCode, count := range result.ErrorCounts { |
| 37 | if errCode < 0 { |
| 38 | c.protocol.Warningf("Internal error %d occurred %d times while parsing channel list from queue manager '%s'", |
| 39 | errCode, count, c.config.QueueManager) |
| 40 | } else { |
| 41 | c.protocol.Warningf("MQ error %d (%s) occurred %d times while parsing channel list from queue manager '%s'", |
| 42 | errCode, mqReasonString(errCode), count, c.config.QueueManager) |
| 43 | } |
| 44 | } |
| 45 | |
| 46 | c.protocol.Debugf("Retrieved %d channels from queue manager '%s'", len(result.Channels), c.config.QueueManager) |
| 47 | |
| 48 | return result.Channels, nil |
| 49 | } |
| 50 | |
| 51 | // GetChannelMetrics returns metrics for a specific channel. |
| 52 | func (c *Client) GetChannelMetrics(channelName string) (*ChannelMetrics, error) { |
| 53 | c.protocol.Debugf("Getting metrics for channel '%s' from queue manager '%s'", channelName, c.config.QueueManager) |
| 54 | |
| 55 | params := []pcfParameter{ |
| 56 | newStringParameter(ibmmq.MQCACH_CHANNEL_NAME, channelName), |
| 57 | } |
| 58 | response, err := c.sendPCFCommand(ibmmq.MQCMD_INQUIRE_CHANNEL_STATUS, params) |
| 59 | if err != nil { |
| 60 | c.protocol.Errorf("Failed to get metrics for channel '%s' from queue manager '%s': %v", |
| 61 | channelName, c.config.QueueManager, err) |
| 62 | return nil, err |
| 63 | } |
| 64 | |
| 65 | attrs, err := c.parsePCFResponseFromParams(response, "") |
| 66 | if err != nil { |
| 67 | c.protocol.Errorf("Failed to parse metrics response for channel '%s' from queue manager '%s': %v", |
| 68 | channelName, c.config.QueueManager, err) |
| 69 | return nil, err |
| 70 | } |
| 71 | |
| 72 | metrics := &ChannelMetrics{ |
| 73 | Name: channelName, |
| 74 | } |
| 75 | |
| 76 | if status, ok := attrs[ibmmq.MQIACH_CHANNEL_STATUS]; ok { |
| 77 | metrics.Status = ChannelStatus(status.(int32)) |
| 78 | } |
| 79 | |
| 80 | if messages, ok := attrs[ibmmq.MQIACH_MSGS]; ok { |
| 81 | val := int64(messages.(int32)) |
| 82 | metrics.Messages = &val |
| 83 | } |
| 84 | if bytes, ok := attrs[ibmmq.MQIACH_BYTES_SENT]; ok { |
| 85 | val := int64(bytes.(int32)) |
| 86 | metrics.Bytes = &val |
| 87 | } |
| 88 | if batches, ok := attrs[ibmmq.MQIACH_BATCHES]; ok { |
| 89 | val := int64(batches.(int32)) |
| 90 | metrics.Batches = &val |
| 91 | } |
| 92 | |
| 93 | // Initialize extended status metrics as NotCollected |
| 94 | metrics.BuffersSent = NotCollected |
| 95 | metrics.BuffersReceived = NotCollected |
| 96 | metrics.CurrentMessages = NotCollected |
| 97 | metrics.XmitQueueTime = NotCollected |
| 98 | metrics.MCAStatus = NotCollected |
| 99 | metrics.InDoubtStatus = NotCollected |
| 100 | metrics.SSLKeyResets = NotCollected |
| 101 | metrics.NPMSpeed = NotCollected |
| 102 | metrics.CurrentSharingConvs = NotCollected |
| 103 | |
| 104 | // Collect extended status metrics if available |
| 105 | if val, ok := attrs[ibmmq.MQIACH_BUFFERS_SENT]; ok { |
| 106 | metrics.BuffersSent = AttributeValue(val.(int32)) |
| 107 | c.protocol.Debugf("Channel '%s' buffers sent: %d", channelName, val.(int32)) |
| 108 | } |
| 109 | if val, ok := attrs[ibmmq.MQIACH_BUFFERS_RCVD]; ok { |
| 110 | metrics.BuffersReceived = AttributeValue(val.(int32)) |
| 111 | c.protocol.Debugf("Channel '%s' buffers received: %d", channelName, val.(int32)) |
| 112 | } |
| 113 | if val, ok := attrs[ibmmq.MQIACH_CURRENT_MSGS]; ok { |
| 114 | metrics.CurrentMessages = AttributeValue(val.(int32)) |
| 115 | c.protocol.Debugf("Channel '%s' current messages: %d", channelName, val.(int32)) |
| 116 | } |
| 117 | if val, ok := attrs[ibmmq.MQIACH_XMITQ_TIME_INDICATOR]; ok { |
| 118 | metrics.XmitQueueTime = AttributeValue(val.(int32)) |
| 119 | c.protocol.Debugf("Channel '%s' XMITQ time: %d", channelName, val.(int32)) |
| 120 | } |
| 121 | if val, ok := attrs[ibmmq.MQIACH_MCA_STATUS]; ok { |
| 122 | metrics.MCAStatus = AttributeValue(val.(int32)) |
| 123 | c.protocol.Debugf("Channel '%s' MCA status: %d", channelName, val.(int32)) |
| 124 | } |
| 125 | if val, ok := attrs[ibmmq.MQIACH_INDOUBT_STATUS]; ok { |
| 126 | metrics.InDoubtStatus = AttributeValue(val.(int32)) |
| 127 | c.protocol.Debugf("Channel '%s' in-doubt status: %d", channelName, val.(int32)) |
| 128 | } |
| 129 | if val, ok := attrs[ibmmq.MQIACH_SSL_KEY_RESETS]; ok { |
| 130 | metrics.SSLKeyResets = AttributeValue(val.(int32)) |
| 131 | c.protocol.Debugf("Channel '%s' SSL key resets: %d", channelName, val.(int32)) |
| 132 | } |
| 133 | if val, ok := attrs[ibmmq.MQIACH_NPM_SPEED]; ok { |
| 134 | metrics.NPMSpeed = AttributeValue(val.(int32)) |
| 135 | c.protocol.Debugf("Channel '%s' NPM speed: %d", channelName, val.(int32)) |
| 136 | } |
| 137 | if val, ok := attrs[ibmmq.MQIACH_CURRENT_SHARING_CONVS]; ok { |
| 138 | metrics.CurrentSharingConvs = AttributeValue(val.(int32)) |
| 139 | c.protocol.Debugf("Channel '%s' current sharing conversations: %d", channelName, val.(int32)) |
| 140 | } |
| 141 | |
| 142 | // Connection name (string attribute) |
| 143 | if val, ok := attrs[ibmmq.MQCACH_CONNECTION_NAME]; ok { |
| 144 | if connStr, ok := val.(string); ok { |
| 145 | metrics.ConnectionName = strings.TrimSpace(connStr) |
| 146 | c.protocol.Debugf("Channel '%s' connection name: '%s'", channelName, metrics.ConnectionName) |
| 147 | } |
| 148 | } |
| 149 | |
| 150 | return metrics, nil |
| 151 | } |
| 152 | |
| 153 | // GetChannelConfig returns configuration for a specific channel. |
| 154 | func (c *Client) GetChannelConfig(channelName string) (*ChannelConfig, error) { |
| 155 | params := []pcfParameter{ |
| 156 | newStringParameter(ibmmq.MQCACH_CHANNEL_NAME, channelName), |
| 157 | } |
| 158 | response, err := c.sendPCFCommand(ibmmq.MQCMD_INQUIRE_CHANNEL, params) |
| 159 | if err != nil { |
| 160 | return nil, err |
| 161 | } |
| 162 | |
| 163 | attrs, err := c.parsePCFResponseFromParams(response, "") |
| 164 | if err != nil { |
| 165 | return nil, err |
| 166 | } |
| 167 | |
| 168 | config := &ChannelConfig{ |
| 169 | Name: channelName, |
| 170 | } |
| 171 | |
| 172 | if channelType, ok := attrs[ibmmq.MQIACH_CHANNEL_TYPE]; ok { |
| 173 | config.Type = ChannelType(channelType.(int32)) |
| 174 | } |
| 175 | |
| 176 | config.BatchSize = NotCollected |
| 177 | config.BatchInterval = NotCollected |
| 178 | config.DiscInterval = NotCollected |
| 179 | config.HbInterval = NotCollected |
| 180 | config.KeepAliveInterval = NotCollected |
| 181 | config.ShortRetry = NotCollected |
| 182 | config.LongRetry = NotCollected |
| 183 | config.MaxMsgLength = NotCollected |
| 184 | config.SharingConversations = NotCollected |
| 185 | config.NetworkPriority = NotCollected |
| 186 | |
| 187 | if batchSize, ok := attrs[ibmmq.MQIACH_BATCH_SIZE]; ok { |
| 188 | config.BatchSize = AttributeValue(batchSize.(int32)) |
| 189 | } |
| 190 | if batchInterval, ok := attrs[ibmmq.MQIACH_BATCH_INTERVAL]; ok { |
| 191 | config.BatchInterval = AttributeValue(batchInterval.(int32)) |
| 192 | } |
| 193 | |
| 194 | if discInterval, ok := attrs[ibmmq.MQIACH_DISC_INTERVAL]; ok { |
| 195 | config.DiscInterval = AttributeValue(discInterval.(int32)) |
| 196 | } |
| 197 | if hbInterval, ok := attrs[ibmmq.MQIACH_HB_INTERVAL]; ok { |
| 198 | config.HbInterval = AttributeValue(hbInterval.(int32)) |
| 199 | } |
| 200 | if keepAliveInterval, ok := attrs[ibmmq.MQIACH_KEEP_ALIVE_INTERVAL]; ok { |
| 201 | config.KeepAliveInterval = AttributeValue(keepAliveInterval.(int32)) |
| 202 | } |
| 203 | |
| 204 | if shortRetry, ok := attrs[ibmmq.MQIACH_SHORT_RETRY]; ok { |
| 205 | config.ShortRetry = AttributeValue(shortRetry.(int32)) |
| 206 | } |
| 207 | if longRetry, ok := attrs[ibmmq.MQIACH_LONG_RETRY]; ok { |
| 208 | config.LongRetry = AttributeValue(longRetry.(int32)) |
| 209 | } |
| 210 | |
| 211 | if maxMsgLength, ok := attrs[ibmmq.MQIACH_MAX_MSG_LENGTH]; ok { |
| 212 | config.MaxMsgLength = AttributeValue(maxMsgLength.(int32)) |
| 213 | } |
| 214 | if sharingConvs, ok := attrs[ibmmq.MQIACH_SHARING_CONVERSATIONS]; ok { |
| 215 | config.SharingConversations = AttributeValue(sharingConvs.(int32)) |
| 216 | } |
| 217 | if netPriority, ok := attrs[ibmmq.MQIACH_NETWORK_PRIORITY]; ok { |
| 218 | config.NetworkPriority = AttributeValue(netPriority.(int32)) |
| 219 | } |
| 220 | |
| 221 | return config, nil |
| 222 | } |
| 223 | |
| 224 | // GetChannels collects comprehensive channel metrics with full transparency statistics |
| 225 | func (c *Client) GetChannels(collectConfig, collectMetrics bool, maxChannels int, selector string, collectSystem bool) (*ChannelCollectionResult, error) { |
| 226 | c.protocol.Debugf("Collecting channel metrics with selector '%s', max=%d, config=%v, metrics=%v, system=%v", |
| 227 | selector, maxChannels, collectConfig, collectMetrics, collectSystem) |
| 228 | |
| 229 | result := &ChannelCollectionResult{ |
| 230 | Stats: CollectionStats{}, |
| 231 | } |
| 232 | |
| 233 | // Step 1: Enhanced Discovery (with channel types) |
| 234 | channelInfos, err := c.discoverChannelsWithInfo(result) |
| 235 | if err != nil { |
| 236 | return result, err |
| 237 | } |
| 238 | |
| 239 | // Step 2: Filtering (including template filtering) |
| 240 | channelInfosToEnrich := c.filterChannelsEnhanced(channelInfos, selector, collectSystem, maxChannels, result) |
| 241 | |
| 242 | // Step 3: Enrichment (now pass channel info with types) |
| 243 | c.enrichChannelsWithTypes(channelInfosToEnrich, collectConfig, collectMetrics, result) |
| 244 | |
| 245 | c.logChannelCollectionSummary(result) |
| 246 | |
| 247 | return result, nil |
| 248 | } |
| 249 | |
| 250 | func (c *Client) discoverChannelsWithInfo(result *ChannelCollectionResult) ([]ChannelInfo, error) { |
| 251 | response, err := c.sendPCFCommand(ibmmq.MQCMD_INQUIRE_CHANNEL, []pcfParameter{ |
| 252 | newStringParameter(ibmmq.MQCACH_CHANNEL_NAME, "*"), |
| 253 | }) |
| 254 | if err != nil { |
| 255 | result.Stats.Discovery.Success = false |
| 256 | c.protocol.Errorf("Channel discovery failed: %v", err) |
| 257 | return nil, fmt.Errorf("channel discovery failed: %w", err) |
| 258 | } |
| 259 | |
| 260 | result.Stats.Discovery.Success = true |
| 261 | |
| 262 | // Parse channel info including type |
| 263 | channelInfos := c.parseChannelInfoFromParams(response) |
| 264 | |
| 265 | successfulItems := int64(len(channelInfos)) |
| 266 | var invisibleItems int64 |
| 267 | // TODO: Add error counting when we update parseChannelInfoFromParams |
| 268 | |
| 269 | result.Stats.Discovery.AvailableItems = successfulItems + invisibleItems |
| 270 | result.Stats.Discovery.InvisibleItems = invisibleItems |
| 271 | |
| 272 | if len(channelInfos) == 0 { |
| 273 | c.protocol.Debugf("No channels discovered") |
| 274 | } else { |
| 275 | // Log channel types for debugging |
| 276 | c.protocol.Debugf("Discovered %d channels with types", len(channelInfos)) |
| 277 | for _, info := range channelInfos { |
| 278 | c.protocol.Debugf(" Channel '%s' type=%d", info.Name, info.Type) |
| 279 | } |
| 280 | } |
| 281 | |
| 282 | return channelInfos, nil |
| 283 | } |
| 284 | |
| 285 | // Keep old function for compatibility |
| 286 | func (c *Client) discoverChannels(result *ChannelCollectionResult) ([]string, error) { |
| 287 | infos, err := c.discoverChannelsWithInfo(result) |
| 288 | if err != nil { |
| 289 | return nil, err |
| 290 | } |
| 291 | |
| 292 | var names []string |
| 293 | for _, info := range infos { |
| 294 | names = append(names, info.Name) |
| 295 | } |
| 296 | return names, nil |
| 297 | } |
| 298 | |
| 299 | func (c *Client) filterChannelsEnhanced(channelInfos []ChannelInfo, selector string, collectSystem bool, maxChannels int, result *ChannelCollectionResult) []ChannelInfo { |
| 300 | visibleItems := result.Stats.Discovery.AvailableItems - result.Stats.Discovery.InvisibleItems |
| 301 | enrichAll := maxChannels <= 0 || visibleItems <= int64(maxChannels) |
| 302 | |
| 303 | c.protocol.Debugf("Discovery found %d visible channels (total: %d, invisible: %d). EnrichAll=%v", |
| 304 | visibleItems, result.Stats.Discovery.AvailableItems, result.Stats.Discovery.InvisibleItems, enrichAll) |
| 305 | |
| 306 | var channelsToEnrich []ChannelInfo |
| 307 | var templateChannelsSkipped int64 |
| 308 | |
| 309 | if enrichAll || selector == "*" { |
| 310 | for _, info := range channelInfos { |
| 311 | // Skip template channels (SYSTEM.DEF.*) |
| 312 | if strings.HasPrefix(info.Name, "SYSTEM.DEF.") { |
| 313 | templateChannelsSkipped++ |
| 314 | c.protocol.Debugf("Skipping template channel '%s' (type=%d) - no runtime status available", |
| 315 | info.Name, info.Type) |
| 316 | continue |
| 317 | } |
| 318 | |
| 319 | if !collectSystem && strings.HasPrefix(info.Name, "SYSTEM.") { |
| 320 | result.Stats.Discovery.ExcludedItems++ |
| 321 | continue |
| 322 | } |
| 323 | channelsToEnrich = append(channelsToEnrich, info) |
| 324 | result.Stats.Discovery.IncludedItems++ |
| 325 | } |
| 326 | c.protocol.Debugf("Enriching %d channels (excluded %d system channels, skipped %d template channels)", |
| 327 | len(channelsToEnrich), result.Stats.Discovery.ExcludedItems, templateChannelsSkipped) |
| 328 | } else { |
| 329 | for _, info := range channelInfos { |
| 330 | // Skip template channels (SYSTEM.DEF.*) |
| 331 | if strings.HasPrefix(info.Name, "SYSTEM.DEF.") { |
| 332 | templateChannelsSkipped++ |
| 333 | c.protocol.Debugf("Skipping template channel '%s' (type=%d) - no runtime status available", |
| 334 | info.Name, info.Type) |
| 335 | continue |
| 336 | } |
| 337 | |
| 338 | if !collectSystem && strings.HasPrefix(info.Name, "SYSTEM.") { |
| 339 | result.Stats.Discovery.ExcludedItems++ |
| 340 | continue |
| 341 | } |
| 342 | |
| 343 | matched, err := filepath.Match(selector, info.Name) |
| 344 | if err != nil { |
| 345 | c.protocol.Warningf("Invalid selector pattern '%s': %v", selector, err) |
| 346 | matched = false |
| 347 | } |
| 348 | |
| 349 | if matched { |
| 350 | channelsToEnrich = append(channelsToEnrich, info) |
| 351 | result.Stats.Discovery.IncludedItems++ |
| 352 | } else { |
| 353 | result.Stats.Discovery.ExcludedItems++ |
| 354 | } |
| 355 | } |
| 356 | c.protocol.Debugf("Selector '%s' matched %d channels, excluded %d (including system filtering), skipped %d templates", |
| 357 | selector, result.Stats.Discovery.IncludedItems, result.Stats.Discovery.ExcludedItems, templateChannelsSkipped) |
| 358 | } |
| 359 | |
| 360 | // Update stats to account for template channels |
| 361 | result.Stats.Discovery.AvailableItems -= templateChannelsSkipped |
| 362 | |
| 363 | return channelsToEnrich |
| 364 | } |
| 365 | |
| 366 | // Keep old function for compatibility |
| 367 | func (c *Client) filterChannels(channels []string, selector string, collectSystem bool, maxChannels int, result *ChannelCollectionResult) []string { |
| 368 | // Convert to ChannelInfo with unknown type for backward compatibility |
| 369 | var infos []ChannelInfo |
| 370 | for _, name := range channels { |
| 371 | infos = append(infos, ChannelInfo{Name: name, Type: 0}) |
| 372 | } |
| 373 | filteredInfos := c.filterChannelsEnhanced(infos, selector, collectSystem, maxChannels, result) |
| 374 | |
| 375 | // Convert back to strings |
| 376 | var names []string |
| 377 | for _, info := range filteredInfos { |
| 378 | names = append(names, info.Name) |
| 379 | } |
| 380 | return names |
| 381 | } |
| 382 | |
| 383 | func (c *Client) enrichChannels(channelsToEnrich []string, collectConfig, collectMetrics bool, result *ChannelCollectionResult) { |
| 384 | for _, channelName := range channelsToEnrich { |
| 385 | cm := ChannelMetrics{ |
| 386 | Name: channelName, |
| 387 | BatchSize: NotCollected, |
| 388 | BatchInterval: NotCollected, |
| 389 | DiscInterval: NotCollected, |
| 390 | HbInterval: NotCollected, |
| 391 | KeepAliveInterval: NotCollected, |
| 392 | ShortRetry: NotCollected, |
| 393 | LongRetry: NotCollected, |
| 394 | MaxMsgLength: NotCollected, |
| 395 | SharingConversations: NotCollected, |
| 396 | NetworkPriority: NotCollected, |
| 397 | } |
| 398 | |
| 399 | if collectConfig { |
| 400 | c.enrichChannelWithConfig(&cm, result) |
| 401 | } |
| 402 | |
| 403 | if collectMetrics { |
| 404 | c.enrichChannelWithMetrics(&cm, result) |
| 405 | } |
| 406 | |
| 407 | result.Channels = append(result.Channels, cm) |
| 408 | } |
| 409 | } |
| 410 | |
| 411 | func (c *Client) enrichChannelsWithTypes(channelInfosToEnrich []ChannelInfo, collectConfig, collectMetrics bool, result *ChannelCollectionResult) { |
| 412 | for _, info := range channelInfosToEnrich { |
| 413 | cm := ChannelMetrics{ |
| 414 | Name: info.Name, |
| 415 | Type: info.Type, // Set type from discovery |
| 416 | BatchSize: NotCollected, |
| 417 | BatchInterval: NotCollected, |
| 418 | DiscInterval: NotCollected, |
| 419 | HbInterval: NotCollected, |
| 420 | KeepAliveInterval: NotCollected, |
| 421 | ShortRetry: NotCollected, |
| 422 | LongRetry: NotCollected, |
| 423 | MaxMsgLength: NotCollected, |
| 424 | SharingConversations: NotCollected, |
| 425 | NetworkPriority: NotCollected, |
| 426 | } |
| 427 | |
| 428 | if collectConfig { |
| 429 | c.enrichChannelWithConfig(&cm, result) |
| 430 | } |
| 431 | |
| 432 | if collectMetrics { |
| 433 | c.enrichChannelWithMetrics(&cm, result) |
| 434 | } |
| 435 | |
| 436 | result.Channels = append(result.Channels, cm) |
| 437 | } |
| 438 | } |
| 439 | |
| 440 | func (c *Client) enrichChannelWithConfig(cm *ChannelMetrics, result *ChannelCollectionResult) { |
| 441 | if result.Stats.Config == nil { |
| 442 | result.Stats.Config = &EnrichmentStats{ |
| 443 | TotalItems: int64(len(result.Channels)), |
| 444 | ErrorCounts: make(map[int32]int), |
| 445 | } |
| 446 | } |
| 447 | |
| 448 | configData, err := c.GetChannelConfig(cm.Name) |
| 449 | if err != nil { |
| 450 | result.Stats.Config.FailedItems++ |
| 451 | if pcfErr, ok := err.(*PCFError); ok { |
| 452 | result.Stats.Config.ErrorCounts[pcfErr.Code]++ |
| 453 | } else { |
| 454 | result.Stats.Config.ErrorCounts[-1]++ |
| 455 | } |
| 456 | c.protocol.Debugf("Failed to get config for channel '%s': %v", cm.Name, err) |
| 457 | } else { |
| 458 | result.Stats.Config.OkItems++ |
| 459 | cm.Type = configData.Type |
| 460 | cm.BatchSize = configData.BatchSize |
| 461 | cm.BatchInterval = configData.BatchInterval |
| 462 | cm.DiscInterval = configData.DiscInterval |
| 463 | cm.HbInterval = configData.HbInterval |
| 464 | cm.KeepAliveInterval = configData.KeepAliveInterval |
| 465 | cm.ShortRetry = configData.ShortRetry |
| 466 | cm.LongRetry = configData.LongRetry |
| 467 | cm.MaxMsgLength = configData.MaxMsgLength |
| 468 | cm.SharingConversations = configData.SharingConversations |
| 469 | cm.NetworkPriority = configData.NetworkPriority |
| 470 | } |
| 471 | } |
| 472 | |
| 473 | func (c *Client) enrichChannelWithMetrics(cm *ChannelMetrics, result *ChannelCollectionResult) { |
| 474 | if result.Stats.Metrics == nil { |
| 475 | result.Stats.Metrics = &EnrichmentStats{ |
| 476 | TotalItems: int64(len(result.Channels)), |
| 477 | ErrorCounts: make(map[int32]int), |
| 478 | } |
| 479 | } |
| 480 | |
| 481 | metricsData, err := c.GetChannelMetrics(cm.Name) |
| 482 | if err != nil { |
| 483 | result.Stats.Metrics.FailedItems++ |
| 484 | if pcfErr, ok := err.(*PCFError); ok { |
| 485 | result.Stats.Metrics.ErrorCounts[pcfErr.Code]++ |
| 486 | } else { |
| 487 | result.Stats.Metrics.ErrorCounts[-1]++ |
| 488 | } |
| 489 | c.protocol.Debugf("Failed to get metrics for channel '%s': %v", cm.Name, err) |
| 490 | } else { |
| 491 | result.Stats.Metrics.OkItems++ |
| 492 | cm.Status = metricsData.Status |
| 493 | cm.Messages = metricsData.Messages |
| 494 | cm.Bytes = metricsData.Bytes |
| 495 | cm.Batches = metricsData.Batches |
| 496 | cm.Connections = metricsData.Connections |
| 497 | cm.BuffersUsed = metricsData.BuffersUsed |
| 498 | cm.BuffersMax = metricsData.BuffersMax |
| 499 | |
| 500 | // Copy extended status metrics |
| 501 | cm.BuffersSent = metricsData.BuffersSent |
| 502 | cm.BuffersReceived = metricsData.BuffersReceived |
| 503 | cm.CurrentMessages = metricsData.CurrentMessages |
| 504 | cm.XmitQueueTime = metricsData.XmitQueueTime |
| 505 | cm.MCAStatus = metricsData.MCAStatus |
| 506 | cm.InDoubtStatus = metricsData.InDoubtStatus |
| 507 | cm.SSLKeyResets = metricsData.SSLKeyResets |
| 508 | cm.NPMSpeed = metricsData.NPMSpeed |
| 509 | cm.CurrentSharingConvs = metricsData.CurrentSharingConvs |
| 510 | cm.ConnectionName = metricsData.ConnectionName |
| 511 | } |
| 512 | } |
| 513 | |
| 514 | func (c *Client) logChannelCollectionSummary(result *ChannelCollectionResult) { |
| 515 | fieldCounts := make(map[string]int) |
| 516 | for _, ch := range result.Channels { |
| 517 | fieldCounts["status"]++ |
| 518 | if ch.Messages != nil { |
| 519 | fieldCounts["messages"]++ |
| 520 | } |
| 521 | if ch.Bytes != nil { |
| 522 | fieldCounts["bytes"]++ |
| 523 | } |
| 524 | if ch.Batches != nil { |
| 525 | fieldCounts["batches"]++ |
| 526 | } |
| 527 | if ch.Connections != nil { |
| 528 | fieldCounts["connections"]++ |
| 529 | } |
| 530 | if ch.BuffersUsed != nil { |
| 531 | fieldCounts["buffers_used"]++ |
| 532 | } |
| 533 | if ch.BuffersMax != nil { |
| 534 | fieldCounts["buffers_max"]++ |
| 535 | } |
| 536 | if ch.BatchSize.IsCollected() { |
| 537 | fieldCounts["batch_size"]++ |
| 538 | } |
| 539 | if ch.BatchInterval.IsCollected() { |
| 540 | fieldCounts["batch_interval"]++ |
| 541 | } |
| 542 | if ch.DiscInterval.IsCollected() { |
| 543 | fieldCounts["disc_interval"]++ |
| 544 | } |
| 545 | if ch.HbInterval.IsCollected() { |
| 546 | fieldCounts["hb_interval"]++ |
| 547 | } |
| 548 | if ch.KeepAliveInterval.IsCollected() { |
| 549 | fieldCounts["keep_alive_interval"]++ |
| 550 | } |
| 551 | if ch.ShortRetry.IsCollected() { |
| 552 | fieldCounts["short_retry"]++ |
| 553 | } |
| 554 | if ch.LongRetry.IsCollected() { |
| 555 | fieldCounts["long_retry"]++ |
| 556 | } |
| 557 | if ch.MaxMsgLength.IsCollected() { |
| 558 | fieldCounts["max_msg_length"]++ |
| 559 | } |
| 560 | if ch.SharingConversations.IsCollected() { |
| 561 | fieldCounts["sharing_conversations"]++ |
| 562 | } |
| 563 | if ch.NetworkPriority.IsCollected() { |
| 564 | fieldCounts["network_priority"]++ |
| 565 | } |
| 566 | } |
| 567 | |
| 568 | c.protocol.Debugf("Channel collection complete - discovered:%d visible:%d included:%d enriched:%d", |
| 569 | result.Stats.Discovery.AvailableItems, |
| 570 | result.Stats.Discovery.AvailableItems-result.Stats.Discovery.InvisibleItems, |
| 571 | result.Stats.Discovery.IncludedItems, |
| 572 | len(result.Channels)) |
| 573 | |
| 574 | c.protocol.Debugf("Channel field collection summary: status=%d messages=%d bytes=%d batches=%d connections=%d "+ |
| 575 | "buffers_used=%d buffers_max=%d batch_size=%d batch_interval=%d disc_interval=%d hb_interval=%d "+ |
| 576 | "keep_alive_interval=%d short_retry=%d long_retry=%d max_msg_length=%d sharing_conversations=%d network_priority=%d", |
| 577 | fieldCounts["status"], fieldCounts["messages"], fieldCounts["bytes"], fieldCounts["batches"], |
| 578 | fieldCounts["connections"], fieldCounts["buffers_used"], fieldCounts["buffers_max"], |
| 579 | fieldCounts["batch_size"], fieldCounts["batch_interval"], fieldCounts["disc_interval"], |
| 580 | fieldCounts["hb_interval"], fieldCounts["keep_alive_interval"], fieldCounts["short_retry"], |
| 581 | fieldCounts["long_retry"], fieldCounts["max_msg_length"], fieldCounts["sharing_conversations"], |
| 582 | fieldCounts["network_priority"]) |
| 583 | } |