| 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 | // GetTopicList returns a list of topics. |
| 16 | func (c *Client) GetTopicList() ([]string, error) { |
| 17 | const pattern = "*" |
| 18 | params := []pcfParameter{ |
| 19 | newStringParameter(ibmmq.MQCA_TOPIC_NAME, pattern), |
| 20 | } |
| 21 | response, err := c.sendPCFCommand(ibmmq.MQCMD_INQUIRE_TOPIC, params) |
| 22 | if err != nil { |
| 23 | return nil, err |
| 24 | } |
| 25 | |
| 26 | result := c.parseTopicListResponseFromParams(response) |
| 27 | |
| 28 | if result.InternalErrors > 0 { |
| 29 | c.protocol.Warningf("encountered %d internal errors while parsing topic list", result.InternalErrors) |
| 30 | } |
| 31 | |
| 32 | for errCode, count := range result.ErrorCounts { |
| 33 | if errCode < 0 { |
| 34 | c.protocol.Debugf("internal error %d occurred %d times", errCode, count) |
| 35 | } else { |
| 36 | c.protocol.Debugf("MQ error %d (%s) occurred %d times", errCode, mqReasonString(errCode), count) |
| 37 | } |
| 38 | } |
| 39 | |
| 40 | return result.Topics, nil |
| 41 | } |
| 42 | |
| 43 | // GetTopicMetrics returns metrics for a specific topic. |
| 44 | func (c *Client) GetTopicMetrics(topicString string) (*TopicMetrics, error) { |
| 45 | c.protocol.Debugf("Getting metrics for topic '%s' from queue manager '%s'", topicString, c.config.QueueManager) |
| 46 | |
| 47 | params := []pcfParameter{ |
| 48 | newStringParameter(ibmmq.MQCA_TOPIC_STRING, topicString), |
| 49 | newIntParameter(ibmmq.MQIACF_TOPIC_STATUS_TYPE, ibmmq.MQIACF_TOPIC_PUB), |
| 50 | } |
| 51 | response, err := c.sendPCFCommand(ibmmq.MQCMD_INQUIRE_TOPIC_STATUS, params) |
| 52 | if err != nil { |
| 53 | c.protocol.Errorf("Failed to get metrics for topic '%s' from queue manager '%s': %v", |
| 54 | topicString, c.config.QueueManager, err) |
| 55 | return nil, err |
| 56 | } |
| 57 | |
| 58 | attrs, err := c.parsePCFResponseFromParams(response, "") |
| 59 | if err != nil { |
| 60 | c.protocol.Errorf("Failed to parse metrics response for topic '%s' from queue manager '%s': %v", |
| 61 | topicString, c.config.QueueManager, err) |
| 62 | return nil, err |
| 63 | } |
| 64 | |
| 65 | metrics := &TopicMetrics{ |
| 66 | TopicString: topicString, |
| 67 | } |
| 68 | |
| 69 | if name, ok := attrs[ibmmq.MQCA_TOPIC_NAME]; ok { |
| 70 | metrics.Name = strings.TrimSpace(name.(string)) |
| 71 | } else { |
| 72 | metrics.Name = topicString |
| 73 | } |
| 74 | |
| 75 | if publishers, ok := attrs[ibmmq.MQIA_PUB_COUNT]; ok { |
| 76 | metrics.Publishers = int64(publishers.(int32)) |
| 77 | } |
| 78 | if subscribers, ok := attrs[ibmmq.MQIA_SUB_COUNT]; ok { |
| 79 | metrics.Subscribers = int64(subscribers.(int32)) |
| 80 | } |
| 81 | |
| 82 | if messages, ok := attrs[ibmmq.MQIAMO_PUBLISH_MSG_COUNT]; ok { |
| 83 | metrics.PublishMsgCount = int64(messages.(int32)) |
| 84 | } |
| 85 | |
| 86 | var lastPubDate, lastPubTime string |
| 87 | if date, ok := attrs[ibmmq.MQCACF_LAST_PUB_DATE]; ok { |
| 88 | lastPubDate = strings.TrimSpace(date.(string)) |
| 89 | } |
| 90 | if time, ok := attrs[ibmmq.MQCACF_LAST_PUB_TIME]; ok { |
| 91 | lastPubTime = strings.TrimSpace(time.(string)) |
| 92 | } |
| 93 | |
| 94 | if lastPubDate != "" && lastPubTime != "" { |
| 95 | // MQCACF_LAST_PUB_DATE/TIME are message-related timestamps, so they should be UTC |
| 96 | if timestamp, err := ParseMQMessageDateTime(lastPubDate, lastPubTime); err == nil { |
| 97 | metrics.LastPubDate = AttributeValue(timestamp.Unix()) |
| 98 | metrics.LastPubTime = AttributeValue(timestamp.Unix()) |
| 99 | } else { |
| 100 | c.protocol.Debugf("Failed to parse last publication timestamp for topic '%s': %v", topicString, err) |
| 101 | } |
| 102 | } |
| 103 | |
| 104 | return metrics, nil |
| 105 | } |
| 106 | |
| 107 | // GetTopics collects comprehensive topic metrics with full transparency statistics |
| 108 | func (c *Client) GetTopics(collectMetrics bool, maxTopics int, selector string, collectSystem bool) (*TopicCollectionResult, error) { |
| 109 | c.protocol.Debugf("Collecting topic metrics with selector '%s', max=%d, metrics=%v, system=%v", |
| 110 | selector, maxTopics, collectMetrics, collectSystem) |
| 111 | |
| 112 | result := &TopicCollectionResult{ |
| 113 | Stats: CollectionStats{}, |
| 114 | } |
| 115 | |
| 116 | // Step 1: Discovery |
| 117 | discoveredTopics, err := c.discoverTopics(result) |
| 118 | if err != nil { |
| 119 | return result, err |
| 120 | } |
| 121 | |
| 122 | // Step 2: Filtering |
| 123 | topicsToEnrich := c.filterTopics(discoveredTopics, selector, collectSystem, maxTopics, result) |
| 124 | |
| 125 | // Step 3: Enrichment |
| 126 | c.enrichTopics(topicsToEnrich, collectMetrics, result) |
| 127 | |
| 128 | c.protocol.Debugf("Topic collection complete - discovered:%d visible:%d included:%d enriched:%d", |
| 129 | result.Stats.Discovery.AvailableItems, |
| 130 | result.Stats.Discovery.AvailableItems-result.Stats.Discovery.InvisibleItems, |
| 131 | result.Stats.Discovery.IncludedItems, |
| 132 | len(result.Topics)) |
| 133 | |
| 134 | return result, nil |
| 135 | } |
| 136 | |
| 137 | func (c *Client) discoverTopics(result *TopicCollectionResult) ([]string, error) { |
| 138 | response, err := c.sendPCFCommand(ibmmq.MQCMD_INQUIRE_TOPIC, []pcfParameter{ |
| 139 | newStringParameter(ibmmq.MQCA_TOPIC_NAME, "*"), |
| 140 | }) |
| 141 | if err != nil { |
| 142 | result.Stats.Discovery.Success = false |
| 143 | c.protocol.Errorf("Topic discovery failed: %v", err) |
| 144 | return nil, fmt.Errorf("topic discovery failed: %w", err) |
| 145 | } |
| 146 | |
| 147 | result.Stats.Discovery.Success = true |
| 148 | parsed := c.parseTopicListResponseFromParams(response) |
| 149 | |
| 150 | successfulItems := int64(len(parsed.Topics)) |
| 151 | var invisibleItems int64 |
| 152 | for _, count := range parsed.ErrorCounts { |
| 153 | invisibleItems += int64(count) |
| 154 | } |
| 155 | |
| 156 | result.Stats.Discovery.AvailableItems = successfulItems + invisibleItems |
| 157 | result.Stats.Discovery.InvisibleItems = invisibleItems |
| 158 | result.Stats.Discovery.ErrorCounts = parsed.ErrorCounts |
| 159 | |
| 160 | for errCode, count := range parsed.ErrorCounts { |
| 161 | if errCode < 0 { |
| 162 | c.protocol.Warningf("Internal error %d occurred %d times during topic discovery", errCode, count) |
| 163 | } else { |
| 164 | c.protocol.Warningf("MQ error %d (%s) occurred %d times during topic discovery", |
| 165 | errCode, mqReasonString(errCode), count) |
| 166 | } |
| 167 | } |
| 168 | |
| 169 | if len(parsed.Topics) == 0 { |
| 170 | c.protocol.Debugf("No topics discovered") |
| 171 | } |
| 172 | |
| 173 | return parsed.Topics, nil |
| 174 | } |
| 175 | |
| 176 | func (c *Client) filterTopics(topics []string, selector string, collectSystem bool, maxTopics int, result *TopicCollectionResult) []string { |
| 177 | visibleItems := result.Stats.Discovery.AvailableItems - result.Stats.Discovery.InvisibleItems |
| 178 | enrichAll := maxTopics <= 0 || visibleItems <= int64(maxTopics) |
| 179 | |
| 180 | c.protocol.Debugf("Discovery found %d visible topics (total: %d, invisible: %d). EnrichAll=%v", |
| 181 | visibleItems, result.Stats.Discovery.AvailableItems, result.Stats.Discovery.InvisibleItems, enrichAll) |
| 182 | |
| 183 | var topicsToEnrich []string |
| 184 | if enrichAll || selector == "*" { |
| 185 | for _, topicName := range topics { |
| 186 | if !collectSystem && strings.HasPrefix(topicName, "SYSTEM.") { |
| 187 | result.Stats.Discovery.ExcludedItems++ |
| 188 | continue |
| 189 | } |
| 190 | topicsToEnrich = append(topicsToEnrich, topicName) |
| 191 | result.Stats.Discovery.IncludedItems++ |
| 192 | } |
| 193 | c.protocol.Debugf("Enriching %d topics (excluded %d system topics)", |
| 194 | len(topicsToEnrich), result.Stats.Discovery.ExcludedItems) |
| 195 | } else { |
| 196 | for _, topicName := range topics { |
| 197 | if !collectSystem && strings.HasPrefix(topicName, "SYSTEM.") { |
| 198 | result.Stats.Discovery.ExcludedItems++ |
| 199 | continue |
| 200 | } |
| 201 | |
| 202 | matched, err := filepath.Match(selector, topicName) |
| 203 | if err != nil { |
| 204 | c.protocol.Warningf("Invalid selector pattern '%s': %v", selector, err) |
| 205 | matched = false |
| 206 | } |
| 207 | |
| 208 | if matched { |
| 209 | topicsToEnrich = append(topicsToEnrich, topicName) |
| 210 | result.Stats.Discovery.IncludedItems++ |
| 211 | } else { |
| 212 | result.Stats.Discovery.ExcludedItems++ |
| 213 | } |
| 214 | } |
| 215 | c.protocol.Debugf("Selector '%s' matched %d topics, excluded %d (including system filtering)", |
| 216 | selector, result.Stats.Discovery.IncludedItems, result.Stats.Discovery.ExcludedItems) |
| 217 | } |
| 218 | return topicsToEnrich |
| 219 | } |
| 220 | |
| 221 | func (c *Client) enrichTopics(topicsToEnrich []string, collectMetrics bool, result *TopicCollectionResult) { |
| 222 | for _, topicName := range topicsToEnrich { |
| 223 | tm := TopicMetrics{TopicString: topicName, Name: topicName} |
| 224 | |
| 225 | if collectMetrics { |
| 226 | c.enrichTopicWithMetrics(&tm, result) |
| 227 | } |
| 228 | |
| 229 | result.Topics = append(result.Topics, tm) |
| 230 | } |
| 231 | } |
| 232 | |
| 233 | func (c *Client) enrichTopicWithMetrics(tm *TopicMetrics, result *TopicCollectionResult) { |
| 234 | if result.Stats.Metrics == nil { |
| 235 | result.Stats.Metrics = &EnrichmentStats{ |
| 236 | TotalItems: int64(len(result.Topics)), |
| 237 | ErrorCounts: make(map[int32]int), |
| 238 | } |
| 239 | } |
| 240 | |
| 241 | metricsData, err := c.GetTopicMetrics(tm.TopicString) |
| 242 | if err != nil { |
| 243 | result.Stats.Metrics.FailedItems++ |
| 244 | if pcfErr, ok := err.(*PCFError); ok { |
| 245 | result.Stats.Metrics.ErrorCounts[pcfErr.Code]++ |
| 246 | } else { |
| 247 | result.Stats.Metrics.ErrorCounts[-1]++ |
| 248 | } |
| 249 | c.protocol.Debugf("Failed to get metrics for topic '%s': %v", tm.TopicString, err) |
| 250 | } else { |
| 251 | result.Stats.Metrics.OkItems++ |
| 252 | tm.Name = metricsData.Name |
| 253 | tm.TopicString = metricsData.TopicString |
| 254 | tm.Publishers = metricsData.Publishers |
| 255 | tm.Subscribers = metricsData.Subscribers |
| 256 | tm.PublishMsgCount = metricsData.PublishMsgCount |
| 257 | tm.LastPubDate = metricsData.LastPubDate |
| 258 | tm.LastPubTime = metricsData.LastPubTime |
| 259 | } |
| 260 | } |