| 1 | package mq |
| 2 | |
| 3 | import ( |
| 4 | "fmt" |
| 5 | "time" |
| 6 | |
| 7 | "github.com/netdata/netdata/go/plugins/plugin/ibm.d/modules/mq/contexts" |
| 8 | ) |
| 9 | |
| 10 | func (c *Collector) collectTopicMetrics() error { |
| 11 | c.Debugf("Collecting topics with selector '%s', system: %v", c.Config.TopicSelector, c.Config.CollectSystemTopics) |
| 12 | |
| 13 | // Use new GetTopics with transparency |
| 14 | result, err := c.client.GetTopics( |
| 15 | true, // collectMetrics (always) |
| 16 | c.Config.MaxTopics, // maxTopics (0 = no limit) |
| 17 | c.Config.TopicSelector, // selector pattern |
| 18 | c.Config.CollectSystemTopics, // collectSystem |
| 19 | ) |
| 20 | if err != nil { |
| 21 | return fmt.Errorf("failed to collect topic metrics: %w", err) |
| 22 | } |
| 23 | |
| 24 | // Check discovery success |
| 25 | if !result.Stats.Discovery.Success { |
| 26 | c.Errorf("Topic discovery failed completely") |
| 27 | return fmt.Errorf("topic discovery failed") |
| 28 | } |
| 29 | |
| 30 | // Map transparency counters to user-facing semantics |
| 31 | monitored := int64(0) |
| 32 | if result.Stats.Metrics != nil { |
| 33 | monitored = result.Stats.Metrics.OkItems |
| 34 | } |
| 35 | |
| 36 | failed := result.Stats.Discovery.UnparsedItems |
| 37 | if result.Stats.Metrics != nil { |
| 38 | failed += result.Stats.Metrics.FailedItems |
| 39 | } |
| 40 | |
| 41 | // Update overview metrics with correct semantics |
| 42 | c.setTopicOverviewMetrics( |
| 43 | monitored, // monitored (successfully enriched) |
| 44 | result.Stats.Discovery.ExcludedItems, // excluded (filtered by user) |
| 45 | result.Stats.Discovery.InvisibleItems, // invisible (discovery errors) |
| 46 | failed, // failed (unparsed + enrichment failures) |
| 47 | ) |
| 48 | |
| 49 | // Log collection summary |
| 50 | c.Debugf("Topic collection complete - discovered:%d visible:%d included:%d collected:%d failed:%d", |
| 51 | result.Stats.Discovery.AvailableItems, |
| 52 | result.Stats.Discovery.AvailableItems-result.Stats.Discovery.InvisibleItems, |
| 53 | result.Stats.Discovery.IncludedItems, |
| 54 | len(result.Topics), |
| 55 | failed) |
| 56 | |
| 57 | // Process collected topic metrics |
| 58 | for _, topic := range result.Topics { |
| 59 | labels := contexts.TopicLabels{ |
| 60 | Topic: topic.TopicString, |
| 61 | } |
| 62 | |
| 63 | // Use structured data from protocol |
| 64 | contexts.Topic.Publishers.Set(c.State, labels, contexts.TopicPublishersValues{ |
| 65 | Publishers: topic.Publishers, |
| 66 | }) |
| 67 | contexts.Topic.Subscribers.Set(c.State, labels, contexts.TopicSubscribersValues{ |
| 68 | Subscribers: topic.Subscribers, |
| 69 | }) |
| 70 | contexts.Topic.Messages.Set(c.State, labels, contexts.TopicMessagesValues{ |
| 71 | Messages: topic.PublishMsgCount, |
| 72 | }) |
| 73 | |
| 74 | // Calculate time since last message if timestamp is available |
| 75 | if topic.LastPubTime > 0 { |
| 76 | currentTime := time.Now().Unix() |
| 77 | timeSinceLastMsg := currentTime - int64(topic.LastPubTime) |
| 78 | |
| 79 | contexts.Topic.TimeSinceLastMessage.Set(c.State, labels, contexts.TopicTimeSinceLastMessageValues{ |
| 80 | Time_since_last_msg: timeSinceLastMsg, |
| 81 | }) |
| 82 | } |
| 83 | } |
| 84 | |
| 85 | return nil |
| 86 | } |