feat(prometheus): Add native /metrics endpoint

activemq-prometheus/README.md

@@ -0,0 +1,92 @@
+<!--
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+-->
+
+# ActiveMQ Prometheus Metrics
+
+## Activation
+
+1. Uncomment the `Prometheus Metrics Web Application` block from `conf/jetty/jetty-webapps.xml`.
+2. Restart the broker
+
+The endpoint uses the existing Jetty management listener, TLS configuration,
+IP allowlist, and JAAS realm. Its path is restricted to the `admins` role
+but can be changed in `conf/jetty/jetty-security.xml`.
+
+## Endpoints
+
+Two endpoints because brokers with many destinations might produce large responses:
+- `GET /metrics`: broker-level metrics only.
+- `GET /metrics?per_object=true`: per-queue, per-topic, and broker-level metrics
+
+## Metrics
+
+### Broker metrics (`activemq_broker_*`)
+
+| Metric | Type | Description |
+|--------|------|-------------|
+| `connections_count` | gauge | Current number of connections |
+| `connections_total` | counter | Total connections since last start |
+| `messages_enqueued_total` | counter | Total messages enqueued since last start |
+| `messages_dequeued_total` | counter | Total messages dequeued since last start |
+| `consumers_count` | gauge | Current number of consumers |
+| `producers_count` | gauge | Current number of producers |
+| `message_count` | gauge | Current number of messages across all destinations |
+| `memory_percent_usage` | gauge | Percent of memory limit used |
+| `memory_limit_bytes` | gauge | Memory limit in bytes |
+| `store_percent_usage` | gauge | Percent of store limit used |
+| `store_limit_bytes` | gauge | Store limit in bytes |
+| `temp_percent_usage` | gauge | Percent of temp limit used |
+| `temp_limit_bytes` | gauge | Temp limit in bytes |
+| `uptime_milliseconds` | gauge | Broker uptime in milliseconds |
+
+### Destination metrics (`activemq_queue_*` / `activemq_topic_*`)
+
+Returned only when `?per_object=true` is set.
+
+| Metric | Type | Description |
+|--------|------|-------------|
+| `message_count` | gauge | Number of messages in destination |
+| `enqueue_count_total` | counter | Total messages enqueued since last start |
+| `dequeue_count_total` | counter | Total messages dequeued since last start |
+| `dispatch_count_total` | counter | Total messages dispatched since last start |
+| `message_inflight_count` | gauge | Messages dispatched but not acknowledged |
+| `expired_count_total` | counter | Total messages expired since last start |
+| `consumer_count` | gauge | Number of consumers |
+| `producer_count` | gauge | Number of producers |
+| `memory_percent_usage` | gauge | Percent of destination memory limit used |
+| `memory_limit_bytes` | gauge | Memory limit for destination in bytes |
+| `memory_usage_bytes` | gauge | Memory used by destination in bytes |
+| `store_message_size_bytes` | gauge | Store message size in bytes |
+| `average_enqueue_time_milliseconds` | gauge | Average time (since last start) messages waited before dispatch |
+
+## Prometheus configuration
+
+Example yaml configuration for running a Prometheus scraper on the same machine as the broker
+```yaml
+scrape_configs:
+  - job_name: activemq
+    metrics_path: /metrics
+    params:
+      per_object: ['true']  # omit for broker-only
+    basic_auth:
+      username: admin
+      password: admin
+    static_configs:
+      - targets: ['localhost:8161']
+```

activemq-prometheus/pom.xml

@@ -0,0 +1,58 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  Licensed to the Apache Software Foundation (ASF) under one or more
+  contributor license agreements.  See the NOTICE file distributed with
+  this work for additional information regarding copyright ownership.
+  The ASF licenses this file to You under the Apache License, Version 2.0
+  (the "License"); you may not use this file except in compliance with
+  the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>org.apache.activemq</groupId>
+        <artifactId>activemq-parent</artifactId>
+        <version>6.4.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>activemq-prometheus</artifactId>
+    <packaging>war</packaging>
+    <name>ActiveMQ :: Prometheus</name>
+    <description>ActiveMQ Prometheus metrics endpoint</description>
+
+    <dependencies>
+
+        <!-- =============================== -->
+        <!-- Required Dependencies -->
+        <!-- =============================== -->
+        <dependency>
+            <groupId>jakarta.servlet</groupId>
+            <artifactId>jakarta.servlet-api</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- =============================== -->
+        <!-- Testing Dependencies            -->
+        <!-- =============================== -->
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <finalName>metrics</finalName>
+    </build>
+
+</project>

activemq-prometheus/src/main/java/org/apache/activemq/prometheus/PrometheusMetricsServlet.java

@@ -0,0 +1,195 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.prometheus;
+
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.lang.management.ManagementFactory;
+import java.util.Set;
+
+import jakarta.servlet.http.HttpServlet;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+
+public class PrometheusMetricsServlet extends HttpServlet {
+
+    private static final long serialVersionUID = 1L;
+    private static final String CONTENT_TYPE = "text/plain; version=0.0.4; charset=utf-8";
+
+    // Metrics can be easily extended by adding them here
+    private static final MetricDefinition[] BROKER_METRICS = {
+        new MetricDefinition("connections", "Current number of connections", "CurrentConnectionsCount", MetricType.GAUGE),
+        new MetricDefinition("connections_total", "Total connections since last start", "TotalConnectionsCount", MetricType.COUNTER),
+        new MetricDefinition("messages_enqueued_total", "Total messages enqueued since last start", "TotalEnqueueCount", MetricType.COUNTER),
+        new MetricDefinition("messages_dequeued_total", "Total messages dequeued since last start", "TotalDequeueCount", MetricType.COUNTER),
+        new MetricDefinition("consumers", "Current number of consumers", "TotalConsumerCount", MetricType.GAUGE),
+        new MetricDefinition("producers", "Current number of producers", "TotalProducerCount", MetricType.GAUGE),
+        new MetricDefinition("messages", "Current number of messages across all destinations", "TotalMessageCount", MetricType.GAUGE),
+        new MetricDefinition("memory_percent_usage", "Percent (0-100) of memory limit used", "MemoryPercentUsage", MetricType.GAUGE),
+        new MetricDefinition("memory_limit_bytes", "Memory limit in bytes", "MemoryLimit", MetricType.GAUGE),
+        new MetricDefinition("store_percent_usage", "Percent (0-100) of store limit used", "StorePercentUsage", MetricType.GAUGE),
+        new MetricDefinition("store_limit_bytes", "Store limit in bytes", "StoreLimit", MetricType.GAUGE),
+        new MetricDefinition("temp_percent_usage", "Percent (0-100) of temp limit used", "TempPercentUsage", MetricType.GAUGE),
+        new MetricDefinition("temp_limit_bytes", "Temp limit in bytes", "TempLimit", MetricType.GAUGE),
+        new MetricDefinition("uptime_milliseconds", "Broker uptime in milliseconds", "UptimeMillis", MetricType.GAUGE)
+    };
+
+    private static final MetricDefinition[] DESTINATION_METRICS = {
+        new MetricDefinition("messages", "Number of messages in this destination", "QueueSize", MetricType.GAUGE),
+        new MetricDefinition("enqueued_total", "Total messages enqueued to this destination since last start", "EnqueueCount", MetricType.COUNTER),
+        new MetricDefinition("dequeued_total", "Total messages dequeued from destination since last start", "DequeueCount", MetricType.COUNTER),
+        new MetricDefinition("dispatched_total", "Total messages dispatched from destination since last start", "DispatchCount", MetricType.COUNTER),
+        new MetricDefinition("message_inflight_count", "Messages dispatched but not acknowledged", "InFlightCount", MetricType.GAUGE),
+        new MetricDefinition("expired_total", "Total messages expired since last start", "ExpiredCount", MetricType.COUNTER),
+        new MetricDefinition("consumers", "Number of consumers", "ConsumerCount", MetricType.GAUGE),
+        new MetricDefinition("producers", "Number of producers", "ProducerCount", MetricType.GAUGE),
+        new MetricDefinition("memory_percent_usage", "Percent (0-100) of destination memory limit used", "MemoryPercentUsage", MetricType.GAUGE),
+        new MetricDefinition("memory_limit_bytes", "Memory limit for this destination in bytes", "MemoryLimit", MetricType.GAUGE),
+        new MetricDefinition("memory_usage_bytes", "Memory used by this destination in bytes", "MemoryUsageByteCount", MetricType.GAUGE),
+        new MetricDefinition("store_message_size_bytes", "Store message size in bytes", "StoreMessageSize", MetricType.GAUGE),
+        new MetricDefinition("average_enqueue_time_milliseconds", "Average time (since last start) messages waited before dispatch", "AverageEnqueueTime", MetricType.GAUGE)
+    };
+
+    @Override
+    protected void doGet(HttpServletRequest request, HttpServletResponse response) throws IOException {
+        boolean perObject = request != null && "true".equalsIgnoreCase(request.getParameter("per_object"));
+
+        StringWriter output = new StringWriter();
+        PrintWriter writer = new PrintWriter(output);
+
+        try {
+            writeMetrics(ManagementFactory.getPlatformMBeanServer(), writer, perObject);
+        } catch (Exception exception) {
+            response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, "Metrics collection failed");
+            return;
+        }
+
+        response.setContentType(CONTENT_TYPE);
+        response.setStatus(HttpServletResponse.SC_OK);
+        response.getWriter().write(output.toString());
+    }
+
+    void writeMetrics(MBeanServer mBeanServer, PrintWriter writer, boolean perObject) throws Exception {
+        writeBrokerMetrics(mBeanServer, writer);
+        // Scraping this by default on brokers with lots of queues or topics might be expensive
+        if (perObject) {
+            writeDestinationMetrics(mBeanServer, writer, "Queue");
+            writeDestinationMetrics(mBeanServer, writer, "Topic");
+        }
+        writer.flush();
+    }
+
+    private void writeBrokerMetrics(MBeanServer mBeanServer, PrintWriter writer) throws Exception {
+        ObjectName pattern = new ObjectName("org.apache.activemq:type=Broker,brokerName=*");
+        Set<ObjectName> brokers = mBeanServer.queryNames(pattern, null);
+
+        for (MetricDefinition metric : BROKER_METRICS) {
+            String metricName = "activemq_broker_" + metric.name;
+            writeMetadata(writer, metricName, metric);
+            for (ObjectName broker : brokers) {
+                String brokerName = sanitizeLabel((String) mBeanServer.getAttribute(broker, "BrokerName"));
+                String labels = "broker=\"" + brokerName + "\"";
+                writeSample(writer, metricName, labels, getNumber(mBeanServer, broker, metric.attribute));
+            }
+        }
+    }
+
+    private void writeDestinationMetrics(MBeanServer mBeanServer, PrintWriter writer, String type) throws Exception {
+        String queryPattern = "org.apache.activemq:type=Broker,brokerName=*,destinationType=" + type + ",destinationName=*";
+        Set<ObjectName> destinations = mBeanServer.queryNames(new ObjectName(queryPattern), null);
+        String typeLower = type.toLowerCase();
+
+        for (MetricDefinition metric : DESTINATION_METRICS) {
+            String metricName = "activemq_" + typeLower + "_" + metric.name;
+            writeMetadata(writer, metricName, metric.withFormattedHelp(typeLower));
+            for (ObjectName destination : destinations) {
+                String brokerName = sanitizeLabel(destination.getKeyProperty("brokerName"));
+                String destinationName = sanitizeLabel(destination.getKeyProperty("destinationName"));
+                String labels = String.format("broker=\"%s\",destination=\"%s\"", brokerName, destinationName);
+                writeSample(writer, metricName, labels, getNumber(mBeanServer, destination, metric.attribute));
+            }
+        }
+    }
+
+    private double getNumber(MBeanServer mBeanServer, ObjectName name, String attribute) {
+        try {
+            Object value = mBeanServer.getAttribute(name, attribute);
+            if (value instanceof Number) {
+                return ((Number) value).doubleValue();
+            }
+        } catch (Exception ignored) {
+            // Some attributes are not available on every ActiveMQ deployment (eg: bridge metrics)
+        }
+        return 0;
+    }
+
+    private void writeMetadata(PrintWriter writer, String metric, MetricDefinition def) {
+        writer.println("# HELP " + metric + " " + def.help);
+        writer.println("# TYPE " + metric + " " + def.type.prometheusName());
+    }
+
+    private void writeSample(PrintWriter writer, String metric, String labels, double value) {
+        if (value == (long) value) {
+            writer.println(metric + "{" + labels + "} " + (long) value);
+        } else {
+            writer.println(metric + "{" + labels + "} " + value);
+        }
+    }
+
+    static String sanitizeLabel(String value) {
+        if (value == null) {
+            return "unknown";
+        }
+        return value.replace("\\", "\\\\").replace("\"", "\\\"").replace("\n", "\\n");
+    }
+
+    private static final class MetricDefinition {
+        private final String name;
+        private final String help;
+        private final String attribute;
+        private final MetricType type;
+
+        private MetricDefinition(String name, String help, String attribute, MetricType type) {
+            this.name = name;
+            this.help = help;
+            this.attribute = attribute;
+            this.type = type;
+        }
+
+        private MetricDefinition withFormattedHelp(String arg) {
+            return new MetricDefinition(name, String.format(help, arg), attribute, type);
+        }
+    }
+
+    enum MetricType {
+        GAUGE("gauge"),
+        COUNTER("counter");
+
+        private final String prometheusName;
+
+        MetricType(String prometheusName) {
+            this.prometheusName = prometheusName;
+        }
+
+        String prometheusName() {
+            return prometheusName;
+        }
+    }
+}