| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package activemq |
| 4 | |
| 5 | import ( |
| 6 | "fmt" |
| 7 | "strings" |
| 8 | ) |
| 9 | |
| 10 | const ( |
| 11 | keyQueues = "queues" |
| 12 | keyTopics = "topics" |
| 13 | keyAdvisory = "Advisory" |
| 14 | ) |
| 15 | |
| 16 | var nameReplacer = strings.NewReplacer(".", "_", " ", "") |
| 17 | |
| 18 | func (c *Collector) collect() (map[string]int64, error) { |
| 19 | metrics := make(map[string]int64) |
| 20 | |
| 21 | var ( |
| 22 | queues *queues |
| 23 | topics *topics |
| 24 | err error |
| 25 | ) |
| 26 | |
| 27 | if queues, err = c.apiClient.getQueues(); err != nil { |
| 28 | return nil, err |
| 29 | } |
| 30 | |
| 31 | if topics, err = c.apiClient.getTopics(); err != nil { |
| 32 | return nil, err |
| 33 | } |
| 34 | |
| 35 | c.processQueues(queues, metrics) |
| 36 | c.processTopics(topics, metrics) |
| 37 | |
| 38 | return metrics, nil |
| 39 | } |
| 40 | |
| 41 | func (c *Collector) processQueues(queues *queues, metrics map[string]int64) { |
| 42 | var ( |
| 43 | count = len(c.activeQueues) |
| 44 | updated = make(map[string]bool) |
| 45 | unp int |
| 46 | ) |
| 47 | |
| 48 | for _, q := range queues.Items { |
| 49 | if strings.Contains(q.Name, keyAdvisory) { |
| 50 | continue |
| 51 | } |
| 52 | |
| 53 | if !c.activeQueues[q.Name] { |
| 54 | if c.MaxQueues != 0 && count > c.MaxQueues { |
| 55 | unp++ |
| 56 | continue |
| 57 | } |
| 58 | |
| 59 | if !c.filterQueues(q.Name) { |
| 60 | continue |
| 61 | } |
| 62 | |
| 63 | c.activeQueues[q.Name] = true |
| 64 | c.addQueueTopicCharts(q.Name, keyQueues) |
| 65 | } |
| 66 | |
| 67 | rname := nameReplacer.Replace(q.Name) |
| 68 | |
| 69 | metrics["queues_"+rname+"_consumers"] = q.Stats.ConsumerCount |
| 70 | metrics["queues_"+rname+"_enqueued"] = q.Stats.EnqueueCount |
| 71 | metrics["queues_"+rname+"_dequeued"] = q.Stats.DequeueCount |
| 72 | metrics["queues_"+rname+"_unprocessed"] = q.Stats.EnqueueCount - q.Stats.DequeueCount |
| 73 | |
| 74 | updated[q.Name] = true |
| 75 | } |
| 76 | |
| 77 | for name := range c.activeQueues { |
| 78 | if !updated[name] { |
| 79 | delete(c.activeQueues, name) |
| 80 | c.removeQueueTopicCharts(name, keyQueues) |
| 81 | } |
| 82 | } |
| 83 | |
| 84 | if unp > 0 { |
| 85 | c.Debugf("%d queues were unprocessed due to max_queues limit (%d)", unp, c.MaxQueues) |
| 86 | } |
| 87 | } |
| 88 | |
| 89 | func (c *Collector) processTopics(topics *topics, metrics map[string]int64) { |
| 90 | var ( |
| 91 | count = len(c.activeTopics) |
| 92 | updated = make(map[string]bool) |
| 93 | unp int |
| 94 | ) |
| 95 | |
| 96 | for _, t := range topics.Items { |
| 97 | if strings.Contains(t.Name, keyAdvisory) { |
| 98 | continue |
| 99 | } |
| 100 | |
| 101 | if !c.activeTopics[t.Name] { |
| 102 | if c.MaxTopics != 0 && count > c.MaxTopics { |
| 103 | unp++ |
| 104 | continue |
| 105 | } |
| 106 | |
| 107 | if !c.filterTopics(t.Name) { |
| 108 | continue |
| 109 | } |
| 110 | |
| 111 | c.activeTopics[t.Name] = true |
| 112 | c.addQueueTopicCharts(t.Name, keyTopics) |
| 113 | } |
| 114 | |
| 115 | rname := nameReplacer.Replace(t.Name) |
| 116 | |
| 117 | metrics["topics_"+rname+"_consumers"] = t.Stats.ConsumerCount |
| 118 | metrics["topics_"+rname+"_enqueued"] = t.Stats.EnqueueCount |
| 119 | metrics["topics_"+rname+"_dequeued"] = t.Stats.DequeueCount |
| 120 | metrics["topics_"+rname+"_unprocessed"] = t.Stats.EnqueueCount - t.Stats.DequeueCount |
| 121 | |
| 122 | updated[t.Name] = true |
| 123 | } |
| 124 | |
| 125 | for name := range c.activeTopics { |
| 126 | if !updated[name] { |
| 127 | // TODO: delete after timeout? |
| 128 | delete(c.activeTopics, name) |
| 129 | c.removeQueueTopicCharts(name, keyTopics) |
| 130 | } |
| 131 | } |
| 132 | |
| 133 | if unp > 0 { |
| 134 | c.Debugf("%d topics were unprocessed due to max_topics limit (%d)", unp, c.MaxTopics) |
| 135 | } |
| 136 | } |
| 137 | |
| 138 | func (c *Collector) filterQueues(line string) bool { |
| 139 | if c.queuesFilter == nil { |
| 140 | return true |
| 141 | } |
| 142 | return c.queuesFilter.MatchString(line) |
| 143 | } |
| 144 | |
| 145 | func (c *Collector) filterTopics(line string) bool { |
| 146 | if c.topicsFilter == nil { |
| 147 | return true |
| 148 | } |
| 149 | return c.topicsFilter.MatchString(line) |
| 150 | } |
| 151 | |
| 152 | func (c *Collector) addQueueTopicCharts(name, typ string) { |
| 153 | rname := nameReplacer.Replace(name) |
| 154 | |
| 155 | charts := charts.Copy() |
| 156 | |
| 157 | for _, chart := range *charts { |
| 158 | chart.ID = fmt.Sprintf(chart.ID, typ, rname) |
| 159 | chart.Title = fmt.Sprintf(chart.Title, name) |
| 160 | chart.Fam = typ |
| 161 | |
| 162 | for _, dim := range chart.Dims { |
| 163 | dim.ID = fmt.Sprintf(dim.ID, typ, rname) |
| 164 | } |
| 165 | } |
| 166 | |
| 167 | _ = c.charts.Add(*charts...) |
| 168 | |
| 169 | } |
| 170 | |
| 171 | func (c *Collector) removeQueueTopicCharts(name, typ string) { |
| 172 | rname := nameReplacer.Replace(name) |
| 173 | |
| 174 | chart := c.charts.Get(fmt.Sprintf("%s_%s_messages", typ, rname)) |
| 175 | chart.MarkRemove() |
| 176 | chart.MarkNotCreated() |
| 177 | |
| 178 | chart = c.charts.Get(fmt.Sprintf("%s_%s_unprocessed_messages", typ, rname)) |
| 179 | chart.MarkRemove() |
| 180 | chart.MarkNotCreated() |
| 181 | |
| 182 | chart = c.charts.Get(fmt.Sprintf("%s_%s_consumers", typ, rname)) |
| 183 | chart.MarkRemove() |
| 184 | chart.MarkNotCreated() |
| 185 | } |