| 1 | package mq |
| 2 | |
| 3 | import ( |
| 4 | "sync" |
| 5 | "time" |
| 6 | |
| 7 | "github.com/netdata/netdata/go/plugins/pkg/matcher" |
| 8 | "github.com/netdata/netdata/go/plugins/plugin/ibm.d/framework" |
| 9 | "github.com/netdata/netdata/go/plugins/plugin/ibm.d/modules/mq/contexts" |
| 10 | "github.com/netdata/netdata/go/plugins/plugin/ibm.d/protocols/pcf" |
| 11 | ) |
| 12 | |
| 13 | // Collector is the collector type. |
| 14 | type Collector struct { |
| 15 | framework.Collector |
| 16 | Config `yaml:",inline" json:",inline"` // Embed config to receive YAML unmarshal |
| 17 | client *pcf.Client |
| 18 | |
| 19 | queueIncludeMatcher matcher.Matcher |
| 20 | queueExcludeMatcher matcher.Matcher |
| 21 | |
| 22 | // Resolved effective intervals (auto-detected or user-configured) |
| 23 | effectiveStatisticsInterval int |
| 24 | effectiveSysTopicInterval int |
| 25 | |
| 26 | // Warning throttling |
| 27 | warnMu sync.Mutex |
| 28 | warns map[string]time.Time |
| 29 | } |
| 30 | |
| 31 | const warnThrottleInterval = 10 * time.Minute |
| 32 | |
| 33 | func (c *Collector) warnOnce(key string, format string, args ...any) { |
| 34 | c.warnMu.Lock() |
| 35 | defer c.warnMu.Unlock() |
| 36 | |
| 37 | if c.warns == nil { |
| 38 | c.warns = make(map[string]time.Time) |
| 39 | } |
| 40 | |
| 41 | now := time.Now() |
| 42 | if last, ok := c.warns[key]; ok && now.Sub(last) < warnThrottleInterval { |
| 43 | return |
| 44 | } |
| 45 | |
| 46 | c.Warningf(format, args...) |
| 47 | c.warns[key] = now |
| 48 | } |
| 49 | |
| 50 | func (c *Collector) clearWarnOnce(key string) { |
| 51 | c.warnMu.Lock() |
| 52 | defer c.warnMu.Unlock() |
| 53 | |
| 54 | if c.warns == nil { |
| 55 | return |
| 56 | } |
| 57 | delete(c.warns, key) |
| 58 | } |
| 59 | |
| 60 | // CollectOnce is called by the framework to collect metrics. |
| 61 | func (c *Collector) CollectOnce() error { |
| 62 | if !c.client.IsConnected() { |
| 63 | if err := c.client.Connect(); err != nil { |
| 64 | return framework.ClassifyError(err, framework.ErrorTemporary) |
| 65 | } |
| 66 | } |
| 67 | |
| 68 | // Get MQ connection info and set as global labels |
| 69 | version, edition, endpoint, err := c.client.GetConnectionInfo() |
| 70 | if err == nil { |
| 71 | c.SetGlobalLabel("version", version) |
| 72 | c.SetGlobalLabel("edition", edition) |
| 73 | c.SetGlobalLabel("endpoint", endpoint) |
| 74 | } else { |
| 75 | c.SetGlobalLabel("version", "unknown") |
| 76 | c.SetGlobalLabel("edition", "unknown") |
| 77 | c.SetGlobalLabel("endpoint", "unknown") |
| 78 | } |
| 79 | |
| 80 | // Collect queue manager metrics |
| 81 | if err := c.collectQueueManagerMetrics(); err != nil { |
| 82 | c.warnOnce("queue_manager_metrics", "failed to collect queue manager metrics: %v", err) |
| 83 | } else { |
| 84 | c.clearWarnOnce("queue_manager_metrics") |
| 85 | } |
| 86 | |
| 87 | // Collect queue metrics |
| 88 | if c.Config.CollectQueues { |
| 89 | if err := c.collectQueueMetrics(); err != nil { |
| 90 | c.warnOnce("queue_metrics", "failed to collect queue metrics: %v", err) |
| 91 | } else { |
| 92 | c.clearWarnOnce("queue_metrics") |
| 93 | } |
| 94 | } |
| 95 | |
| 96 | // Collect channel metrics |
| 97 | if c.Config.CollectChannels { |
| 98 | if err := c.collectChannelMetrics(); err != nil { |
| 99 | c.warnOnce("channel_metrics", "failed to collect channel metrics: %v", err) |
| 100 | } else { |
| 101 | c.clearWarnOnce("channel_metrics") |
| 102 | } |
| 103 | } |
| 104 | |
| 105 | // Collect topic metrics |
| 106 | if c.Config.CollectTopics { |
| 107 | if err := c.collectTopicMetrics(); err != nil { |
| 108 | c.warnOnce("topic_metrics", "failed to collect topic metrics: %v", err) |
| 109 | } else { |
| 110 | c.clearWarnOnce("topic_metrics") |
| 111 | } |
| 112 | } |
| 113 | |
| 114 | // Collect listener metrics |
| 115 | if c.Config.CollectListeners { |
| 116 | if err := c.collectListenerMetrics(); err != nil { |
| 117 | c.warnOnce("listener_metrics", "failed to collect listener metrics: %v", err) |
| 118 | } else { |
| 119 | c.clearWarnOnce("listener_metrics") |
| 120 | } |
| 121 | } |
| 122 | |
| 123 | // Collect subscription metrics |
| 124 | if c.Config.CollectSubscriptions { |
| 125 | if err := c.collectSubscriptions(); err != nil { |
| 126 | c.warnOnce("subscription_metrics", "failed to collect subscription metrics: %v", err) |
| 127 | } else { |
| 128 | c.clearWarnOnce("subscription_metrics") |
| 129 | } |
| 130 | } |
| 131 | |
| 132 | // Collect statistics queue metrics (advanced metrics) - every iteration |
| 133 | if c.Config.CollectStatisticsQueue { |
| 134 | if err := c.collectStatistics(); err != nil { |
| 135 | c.warnOnce("statistics_queue", "failed to collect statistics queue metrics: %v", err) |
| 136 | } else { |
| 137 | c.clearWarnOnce("statistics_queue") |
| 138 | } |
| 139 | } |
| 140 | |
| 141 | // Collect $SYS topic metrics (resource metrics) - every iteration |
| 142 | if c.Config.CollectSysTopics { |
| 143 | if err := c.collectSysTopics(); err != nil { |
| 144 | c.Warningf("failed to collect $SYS topic metrics: %v", err) |
| 145 | } |
| 146 | } |
| 147 | |
| 148 | return nil |
| 149 | } |
| 150 | |
| 151 | func (c *Collector) collectQueueManagerMetrics() error { |
| 152 | metrics, err := c.client.GetQueueManagerStatus() |
| 153 | if err != nil { |
| 154 | return err |
| 155 | } |
| 156 | |
| 157 | contexts.QueueManager.Status.Set(c.State, contexts.EmptyLabels{}, contexts.QueueManagerStatusValues{ |
| 158 | Status: metrics.Status, |
| 159 | }) |
| 160 | |
| 161 | // Collect connection count if available |
| 162 | if metrics.ConnectionCount.IsCollected() { |
| 163 | contexts.QueueManager.ConnectionCount.Set(c.State, contexts.EmptyLabels{}, contexts.QueueManagerConnectionCountValues{ |
| 164 | Connections: metrics.ConnectionCount.Int64(), |
| 165 | }) |
| 166 | } |
| 167 | |
| 168 | // Collect uptime if available |
| 169 | if metrics.Uptime.IsCollected() { |
| 170 | contexts.QueueManager.Uptime.Set(c.State, contexts.EmptyLabels{}, contexts.QueueManagerUptimeValues{ |
| 171 | Uptime: metrics.Uptime.Int64(), |
| 172 | }) |
| 173 | } |
| 174 | |
| 175 | return nil |
| 176 | } |
| 177 | |
| 178 | // GetEffectiveStatisticsInterval returns the resolved statistics interval |
| 179 | // (auto-detected from STATINT or user-configured value) |
| 180 | func (c *Collector) GetEffectiveStatisticsInterval() int { |
| 181 | return c.effectiveStatisticsInterval |
| 182 | } |
| 183 | |
| 184 | // GetEffectiveSysTopicInterval returns the resolved $SYS topic interval |
| 185 | // (auto-detected from MONINT or user-configured value) |
| 186 | func (c *Collector) GetEffectiveSysTopicInterval() int { |
| 187 | return c.effectiveSysTopicInterval |
| 188 | } |
| 189 | |
| 190 | // resolveIntervals determines the effective intervals to use |
| 191 | func (c *Collector) resolveIntervals() { |
| 192 | // Statistics interval: auto-detected STATINT overwrites user configuration |
| 193 | if autoDetected := c.client.GetStatisticsInterval(); autoDetected > 0 { |
| 194 | c.effectiveStatisticsInterval = int(autoDetected) |
| 195 | c.Infof("Using auto-detected statistics interval from STATINT: %d seconds (overwrites config)", c.effectiveStatisticsInterval) |
| 196 | } else { |
| 197 | // Use configured value (default 60s) |
| 198 | c.effectiveStatisticsInterval = c.Config.StatisticsInterval |
| 199 | c.Infof("Using configured statistics interval: %d seconds (STATINT not available)", c.effectiveStatisticsInterval) |
| 200 | } |
| 201 | |
| 202 | // $SYS topic interval: use configured value (default 10s, user can override) |
| 203 | c.effectiveSysTopicInterval = c.Config.SysTopicInterval |
| 204 | c.Infof("Using configured $SYS topic interval: %d seconds", c.effectiveSysTopicInterval) |
| 205 | } |