| 1 | package mq |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "fmt" |
| 7 | "strings" |
| 8 | |
| 9 | "github.com/netdata/netdata/go/plugins/pkg/matcher" |
| 10 | "github.com/netdata/netdata/go/plugins/plugin/ibm.d/modules/mq/contexts" |
| 11 | "github.com/netdata/netdata/go/plugins/plugin/ibm.d/protocols/pcf" |
| 12 | ) |
| 13 | |
| 14 | func compileMatcher(patterns []string) (matcher.Matcher, error) { |
| 15 | if len(patterns) == 0 { |
| 16 | return nil, nil |
| 17 | } |
| 18 | |
| 19 | expr := strings.Join(patterns, " ") |
| 20 | expr = strings.TrimSpace(expr) |
| 21 | if expr == "" { |
| 22 | return nil, nil |
| 23 | } |
| 24 | |
| 25 | return matcher.NewSimplePatternsMatcher(expr) |
| 26 | } |
| 27 | |
| 28 | // defaultConfig returns a new Config with all default values set. |
| 29 | // This is used both for New() and for the module registration to ensure |
| 30 | // consistency and single source of truth for defaults. |
| 31 | func defaultConfig() Config { |
| 32 | return Config{ |
| 33 | // Connection defaults |
| 34 | QueueManager: "QM1", |
| 35 | Host: "localhost", |
| 36 | Port: 1414, |
| 37 | Channel: "SYSTEM.DEF.SVRCONN", |
| 38 | User: "", // No authentication by default |
| 39 | Password: "", // No authentication by default |
| 40 | |
| 41 | // Collection defaults - collect everything by default |
| 42 | CollectQueues: true, |
| 43 | CollectChannels: true, |
| 44 | CollectTopics: true, |
| 45 | CollectListeners: true, |
| 46 | CollectSubscriptions: true, |
| 47 | CollectSystemQueues: true, |
| 48 | CollectSystemChannels: true, |
| 49 | CollectSystemTopics: true, |
| 50 | CollectSystemListeners: true, |
| 51 | CollectChannelConfig: true, |
| 52 | CollectQueueConfig: true, |
| 53 | |
| 54 | // Only destructive operations are disabled by default |
| 55 | CollectResetQueueStats: false, |
| 56 | // Statistics queue collection is disabled by default (may not be available on all systems) |
| 57 | CollectStatisticsQueue: false, |
| 58 | // $SYS topic collection is disabled by default (requires MQ 9.0+) |
| 59 | CollectSysTopics: false, |
| 60 | |
| 61 | // Interval defaults |
| 62 | StatisticsInterval: 60, // Default 60s, auto-detected STATINT overwrites if available |
| 63 | SysTopicInterval: 10, // Default 10s per IBM docs, user can override if customized |
| 64 | |
| 65 | // Queue selection defaults: monitor core system queues, ignore churny patterns |
| 66 | IncludeQueues: []string{ |
| 67 | "SYSTEM.DEAD.LETTER.QUEUE", |
| 68 | "SYSTEM.ADMIN.COMMAND.QUEUE", |
| 69 | "SYSTEM.ADMIN.STATISTICS.QUEUE", |
| 70 | }, |
| 71 | ExcludeQueues: []string{ |
| 72 | "SYSTEM.*", |
| 73 | "AMQ.*", |
| 74 | }, |
| 75 | |
| 76 | ChannelSelector: "", |
| 77 | TopicSelector: "", |
| 78 | ListenerSelector: "", |
| 79 | SubscriptionSelector: "", |
| 80 | |
| 81 | // Cardinality control defaults |
| 82 | MaxQueues: 50, |
| 83 | MaxChannels: 100, |
| 84 | MaxTopics: 100, |
| 85 | MaxListeners: 100, |
| 86 | } |
| 87 | } |
| 88 | |
| 89 | func (c *Collector) Init(ctx context.Context) error { |
| 90 | // Set this collector as the implementation |
| 91 | c.SetImpl(c) |
| 92 | |
| 93 | // Copy framework configuration from module config to framework |
| 94 | // Only if user provided values (non-zero) |
| 95 | if c.Config.ObsoletionIterations != 0 { |
| 96 | c.Collector.Config.ObsoletionIterations = c.Config.ObsoletionIterations |
| 97 | } |
| 98 | if c.Config.UpdateEvery != 0 { |
| 99 | c.Collector.Config.UpdateEvery = c.Config.UpdateEvery |
| 100 | } |
| 101 | if c.Config.CollectionGroups != nil { |
| 102 | c.Collector.Config.CollectionGroups = c.Config.CollectionGroups |
| 103 | } |
| 104 | |
| 105 | // Register all contexts from generated code BEFORE base init |
| 106 | c.RegisterContexts(contexts.GetAllContexts()...) |
| 107 | |
| 108 | // Initialize framework |
| 109 | if err := c.Collector.Init(ctx); err != nil { |
| 110 | return err |
| 111 | } |
| 112 | |
| 113 | // Statistics contexts will be configured with appropriate update interval |
| 114 | // when they are first collected during statistics collection |
| 115 | if c.Config.CollectStatisticsQueue { |
| 116 | c.Infof("Statistics collection enabled - interval will be determined during Check()") |
| 117 | } |
| 118 | |
| 119 | // Debug log the complete configuration as JSON |
| 120 | if configJSON, err := json.Marshal(c.Config); err == nil { |
| 121 | c.Debugf("Running with configuration: %s", string(configJSON)) |
| 122 | } |
| 123 | |
| 124 | // Create client |
| 125 | c.client = pcf.NewClient(pcf.Config{ |
| 126 | QueueManager: c.Config.QueueManager, |
| 127 | Channel: c.Config.Channel, |
| 128 | Host: c.Config.Host, |
| 129 | Port: c.Config.Port, |
| 130 | User: c.Config.User, |
| 131 | Password: c.Config.Password, |
| 132 | }, c.State) |
| 133 | |
| 134 | includeMatcher, err := compileMatcher(c.Config.IncludeQueues) |
| 135 | if err != nil { |
| 136 | return fmt.Errorf("invalid include_queues patterns: %w", err) |
| 137 | } |
| 138 | excludeMatcher, err := compileMatcher(c.Config.ExcludeQueues) |
| 139 | if err != nil { |
| 140 | return fmt.Errorf("invalid exclude_queues patterns: %w", err) |
| 141 | } |
| 142 | c.queueIncludeMatcher = includeMatcher |
| 143 | c.queueExcludeMatcher = excludeMatcher |
| 144 | |
| 145 | if len(c.Config.IncludeQueues) == 0 { |
| 146 | c.Infof("Queue include patterns: none (all queues eligible)") |
| 147 | } else { |
| 148 | c.Infof("Queue include patterns: %v", c.Config.IncludeQueues) |
| 149 | } |
| 150 | if len(c.Config.ExcludeQueues) == 0 { |
| 151 | c.Infof("Queue exclude patterns: none") |
| 152 | } else { |
| 153 | c.Infof("Queue exclude patterns: %v", c.Config.ExcludeQueues) |
| 154 | } |
| 155 | |
| 156 | if c.Config.ChannelSelector == "" { |
| 157 | c.Infof("Channel selector: empty (no channels will be collected)") |
| 158 | } else if c.Config.ChannelSelector == "*" { |
| 159 | c.Infof("Channel selector: all channels will be collected") |
| 160 | } else { |
| 161 | c.Infof("Channel selector configured: %s (applied after discovery)", c.Config.ChannelSelector) |
| 162 | } |
| 163 | |
| 164 | // Warn about destructive statistics collection |
| 165 | if c.Config.CollectResetQueueStats { |
| 166 | c.Warningf("DESTRUCTIVE statistics collection is ENABLED!") |
| 167 | c.Warningf("Queue message counters will be RESET TO ZERO after each collection!") |
| 168 | c.Warningf("This WILL BREAK other monitoring tools using the same statistics!") |
| 169 | c.Warningf("Only use this if Netdata is the ONLY monitoring tool for MQ!") |
| 170 | } |
| 171 | |
| 172 | // Log statistics queue collection status |
| 173 | if c.Config.CollectStatisticsQueue { |
| 174 | c.Infof("Statistics queue collection is ENABLED") |
| 175 | c.Infof("Will collect advanced metrics from SYSTEM.ADMIN.STATISTICS.QUEUE") |
| 176 | c.Infof("Note: Statistics must be enabled on the queue manager (STATQ, STATINT settings)") |
| 177 | } |
| 178 | |
| 179 | // Log $SYS topic collection status |
| 180 | if c.Config.CollectSysTopics { |
| 181 | c.Infof("$SYS topic collection is ENABLED") |
| 182 | c.Infof("Will collect resource metrics (CPU, memory, log) from $SYS topics") |
| 183 | c.Infof("Note: Requires MQ 9.0+ with MONINT configured") |
| 184 | } |
| 185 | |
| 186 | return nil |
| 187 | } |
| 188 | |
| 189 | func (c *Collector) Check(ctx context.Context) error { |
| 190 | // Try to connect |
| 191 | if err := c.client.Connect(); err != nil { |
| 192 | return fmt.Errorf("connection check failed: %w", err) |
| 193 | } |
| 194 | |
| 195 | // Auto-detect intervals from queue manager configuration |
| 196 | c.resolveIntervals() |
| 197 | |
| 198 | return nil |
| 199 | } |
| 200 | |
| 201 | func (c *Collector) Cleanup(ctx context.Context) { |
| 202 | if c.client != nil { |
| 203 | c.client.Disconnect() |
| 204 | } |
| 205 | } |