This is an automated email from the ASF dual-hosted git repository.
mattisonchao pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.9 by this push:
new f7058ba58be [branch-2.9][enh][monitor]: add metrics for pulsar web
service thread pool (#15741)
f7058ba58be is described below
commit f7058ba58bef010c5eb9776c0856bbdc231fda24
Author: Qiang Zhao <[email protected]>
AuthorDate: Wed May 25 07:26:14 2022 +0800
[branch-2.9][enh][monitor]: add metrics for pulsar web service thread pool
(#15741)
---
.../prometheus/PrometheusMetricsGenerator.java | 4 +-
.../apache/pulsar/broker/web/WebExecutorStats.java | 100 +++++++++++++++++++++
.../org/apache/pulsar/broker/web/WebService.java | 4 +
.../pulsar/broker/stats/PrometheusMetricsTest.java | 6 +-
.../apache/pulsar/broker/web/WebServiceTest.java | 47 ++++++++--
5 files changed, 151 insertions(+), 10 deletions(-)
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java
index 9d5e1c77c69..cd6afd1535d 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java
@@ -242,7 +242,9 @@ public class PrometheusMetricsGenerator {
for (int i = 0; i < metricFamily.samples.size(); i++) {
Sample sample = metricFamily.samples.get(i);
stream.write(sample.name);
- stream.write("{cluster=\"").write(cluster).write('"');
+ if (!sample.labelNames.contains("cluster")) {
+ stream.write("{cluster=\"").write(cluster).write('"');
+ }
for (int j = 0; j < sample.labelNames.size(); j++) {
String labelValue = sample.labelValues.get(j);
if (labelValue != null) {
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebExecutorStats.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebExecutorStats.java
new file mode 100644
index 00000000000..45f3a1e562b
--- /dev/null
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebExecutorStats.java
@@ -0,0 +1,100 @@
+/**
+ * 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.pulsar.broker.web;
+
+import io.prometheus.client.CollectorRegistry;
+import io.prometheus.client.Gauge;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class WebExecutorStats implements AutoCloseable {
+ private static final AtomicBoolean CLOSED = new AtomicBoolean(false);
+
+ private final Gauge maxThreads;
+ private final Gauge minThreads;
+ private final Gauge idleThreads;
+ private final Gauge activeThreads;
+ private final Gauge currentThreads;
+ private final WebExecutorThreadPool executor;
+
+ private static volatile WebExecutorStats instance;
+
+ static synchronized WebExecutorStats getStats(WebExecutorThreadPool
executor) {
+ if (null == instance) {
+ instance = new WebExecutorStats(executor);
+ }
+
+ return instance;
+ }
+
+ private WebExecutorStats(WebExecutorThreadPool executor) {
+ this.executor = executor;
+
+ this.maxThreads = Gauge.build("pulsar_web_executor_max_threads",
"-").create()
+ .setChild(new Gauge.Child() {
+ public double get() {
+ return WebExecutorStats.this.executor.getMaxThreads();
+ }
+ })
+ .register();
+
+ this.minThreads = Gauge.build("pulsar_web_executor_min_threads",
"-").create()
+ .setChild(new Gauge.Child() {
+ public double get() {
+ return WebExecutorStats.this.executor.getMinThreads();
+ }
+ })
+ .register();
+
+ this.idleThreads = Gauge.build("pulsar_web_executor_idle_threads",
"-").create()
+ .setChild(new Gauge.Child() {
+ public double get() {
+ return WebExecutorStats.this.executor.getIdleThreads();
+ }
+ })
+ .register();
+
+ this.activeThreads = Gauge.build("pulsar_web_executor_active_threads",
"-").create()
+ .setChild(new Gauge.Child() {
+ public double get() {
+ return WebExecutorStats.this.executor.getThreads()
+ -
WebExecutorStats.this.executor.getIdleThreads();
+ }
+ })
+ .register();
+
+ this.currentThreads =
Gauge.build("pulsar_web_executor_current_threads", "-").create()
+ .setChild(new Gauge.Child() {
+ public double get() {
+ return WebExecutorStats.this.executor.getThreads();
+ }
+ })
+ .register();
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (CLOSED.compareAndSet(false, true)) {
+ CollectorRegistry.defaultRegistry.unregister(this.activeThreads);
+ CollectorRegistry.defaultRegistry.unregister(this.maxThreads);
+ CollectorRegistry.defaultRegistry.unregister(this.minThreads);
+ CollectorRegistry.defaultRegistry.unregister(this.idleThreads);
+ CollectorRegistry.defaultRegistry.unregister(this.currentThreads);
+ }
+ }
+}
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
index e80a5160cb5..f2542745bfa 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
@@ -65,6 +65,8 @@ public class WebService implements AutoCloseable {
private final PulsarService pulsar;
private final Server server;
private final List<Handler> handlers;
+
+ private final WebExecutorStats executorStats;
private final WebExecutorThreadPool webServiceExecutor;
public final int maxConcurrentRequests;
@@ -78,6 +80,7 @@ public class WebService implements AutoCloseable {
this.webServiceExecutor = new WebExecutorThreadPool(
pulsar.getConfiguration().getNumHttpServerThreads(),
"pulsar-web");
+ this.executorStats = WebExecutorStats.getStats(webServiceExecutor);
this.server = new Server(webServiceExecutor);
this.maxConcurrentRequests =
pulsar.getConfiguration().getMaxConcurrentHttpRequests();
List<ServerConnector> connectors = new ArrayList<>();
@@ -275,6 +278,7 @@ public class WebService implements AutoCloseable {
jettyStatisticsCollector = null;
}
webServiceExecutor.join();
+ this.executorStats.close();
log.info("Web service closed");
} catch (Exception e) {
throw new PulsarServerException(e);
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
index 3a7d23b06e7..b4094e488f3 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
@@ -1317,9 +1317,9 @@ public class PrometheusMetricsTest extends BrokerTestBase
{
return parsed;
}
- static class Metric {
- Map<String, String> tags = new TreeMap<>();
- double value;
+ public static class Metric {
+ public Map<String, String> tags = new TreeMap<>();
+ public double value;
@Override
public String toString() {
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceTest.java
index b8f3ac472c8..23a3db91692 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceTest.java
@@ -24,12 +24,14 @@ import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
+import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import com.google.common.io.CharStreams;
import com.google.common.io.Closeables;
import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
@@ -38,12 +40,7 @@ import java.security.KeyStore;
import java.security.PrivateKey;
import java.security.SecureRandom;
import java.security.cert.Certificate;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
+import java.util.*;
import java.util.concurrent.CompletableFuture;
import javax.net.ssl.HttpsURLConnection;
@@ -59,6 +56,8 @@ import org.apache.pulsar.broker.MockedBookKeeperClientFactory;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.stats.PrometheusMetricsTest;
+import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsGenerator;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.admin.PulsarAdminException.ConflictException;
@@ -99,6 +98,42 @@ public class WebServiceTest {
private static final String TLS_CLIENT_CERT_FILE_PATH =
"./src/test/resources/certificate/client.crt";
private static final String TLS_CLIENT_KEY_FILE_PATH =
"./src/test/resources/certificate/client.key";
+ @Test
+ public void testWebExecutorMetrics() throws Exception {
+ setupEnv(true, "1.0", true, false, false, false, -1, false);
+ ByteArrayOutputStream statsOut = new ByteArrayOutputStream();
+ PrometheusMetricsGenerator.generate(pulsar, false, false, false,
statsOut);
+ String metricsStr = statsOut.toString();
+ Multimap<String, PrometheusMetricsTest.Metric> metrics =
PrometheusMetricsTest.parseMetrics(metricsStr);
+
+ Collection<PrometheusMetricsTest.Metric> maxThreads =
metrics.get("pulsar_web_executor_max_threads");
+ Collection<PrometheusMetricsTest.Metric> minThreads =
metrics.get("pulsar_web_executor_min_threads");
+ Collection<PrometheusMetricsTest.Metric> activeThreads =
metrics.get("pulsar_web_executor_active_threads");
+ Collection<PrometheusMetricsTest.Metric> idleThreads =
metrics.get("pulsar_web_executor_idle_threads");
+ Collection<PrometheusMetricsTest.Metric> currentThreads =
metrics.get("pulsar_web_executor_current_threads");
+
+ for (PrometheusMetricsTest.Metric metric : maxThreads) {
+ Assert.assertNotNull(metric.tags.get("cluster"));
+ Assert.assertTrue(metric.value > 0);
+ }
+ for (PrometheusMetricsTest.Metric metric : minThreads) {
+ Assert.assertNotNull(metric.tags.get("cluster"));
+ Assert.assertTrue(metric.value > 0);
+ }
+ for (PrometheusMetricsTest.Metric metric : activeThreads) {
+ Assert.assertNotNull(metric.tags.get("cluster"));
+ Assert.assertTrue(metric.value >= 0);
+ }
+ for (PrometheusMetricsTest.Metric metric : idleThreads) {
+ Assert.assertNotNull(metric.tags.get("cluster"));
+ Assert.assertTrue(metric.value >= 0);
+ }
+ for (PrometheusMetricsTest.Metric metric : currentThreads) {
+ Assert.assertNotNull(metric.tags.get("cluster"));
+ Assert.assertTrue(metric.value > 0);
+ }
+ }
+
/**
* Test that the {@WebService} class properly passes the
allowUnversionedClients value. We do this by setting
* allowUnversionedClients to true, then making a request with no version,
which should go through.
