You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2018/10/03 18:15:40 UTC
[pulsar] branch master updated: [functions][metrics] Update
function metrics (#2686)
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new e31c2d8 [functions][metrics] Update function metrics (#2686)
e31c2d8 is described below
commit e31c2d8e0885ba9490c7f2b2acdafd26996ded23
Author: Sijie Guo <gu...@gmail.com>
AuthorDate: Wed Oct 3 11:15:33 2018 -0700
[functions][metrics] Update function metrics (#2686)
* Fix stats generator
* Set the response to text/plain
* fix TopicStats and NamespaceStats
---
.../stats/prometheus/NamespaceStatsAggregator.java | 25 ++-------------
.../pulsar/broker/stats/prometheus/TopicStats.java | 36 +++-------------------
.../functions/worker/FunctionsStatsGenerator.java | 10 +-----
.../worker/rest/api/FunctionsMetricsResource.java | 5 ++-
4 files changed, 13 insertions(+), 63 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java
index 36343cf..8a727f2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java
@@ -24,12 +24,9 @@ import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.common.policies.data.ReplicatorStats;
import org.apache.pulsar.common.util.SimpleTextOutputStream;
-import org.eclipse.jetty.util.ConcurrentHashSet;
import io.netty.util.concurrent.FastThreadLocal;
-import java.util.Set;
-
public class NamespaceStatsAggregator {
private static FastThreadLocal<AggregatedNamespaceStats> localNamespaceStats = new FastThreadLocal<AggregatedNamespaceStats>() {
@@ -46,8 +43,6 @@ public class NamespaceStatsAggregator {
}
};
- private static Set<String> METRIC_TYPES = new ConcurrentHashSet<>();
-
public static void generate(PulsarService pulsar, boolean includeTopicMetrics, boolean includeConsumerMetrics, SimpleTextOutputStream stream) {
String cluster = pulsar.getConfiguration().getClusterName();
AggregatedNamespaceStats namespaceStats = localNamespaceStats.get();
@@ -56,8 +51,6 @@ public class NamespaceStatsAggregator {
pulsar.getBrokerService().getMultiLayerTopicMap().forEach((namespace, bundlesMap) -> {
namespaceStats.reset();
- METRIC_TYPES.forEach(metric -> TopicStats.metricType(stream, metric));
-
bundlesMap.forEach((bundle, topicsMap) -> {
topicsMap.forEach((name, topic) -> {
getTopicStats(topic, topicStats, includeConsumerMetrics);
@@ -225,33 +218,21 @@ public class NamespaceStatsAggregator {
private static void metric(SimpleTextOutputStream stream, String cluster, String namespace, String name,
long value) {
- if (!METRIC_TYPES.contains(name)) {
- TopicStats.metricType(stream, name);
- METRIC_TYPES.add(name);
- }
-
+ TopicStats.metricType(stream, name);
stream.write(name).write("{cluster=\"").write(cluster).write("\",namespace=\"").write(namespace).write("\"} ");
stream.write(value).write(' ').write(System.currentTimeMillis()).write('\n');
}
private static void metric(SimpleTextOutputStream stream, String cluster, String namespace, String name,
double value) {
- if (!METRIC_TYPES.contains(name)) {
- TopicStats.metricType(stream, name);
- METRIC_TYPES.add(name);
- }
-
+ TopicStats.metricType(stream, name);
stream.write(name).write("{cluster=\"").write(cluster).write("\",namespace=\"").write(namespace).write("\"} ");
stream.write(value).write(' ').write(System.currentTimeMillis()).write('\n');
}
private static void metricWithRemoteCluster(SimpleTextOutputStream stream, String cluster, String namespace,
String name, String remoteCluster, double value) {
- if (!METRIC_TYPES.contains(name)) {
- TopicStats.metricType(stream, name);
- METRIC_TYPES.add(name);
- }
-
+ TopicStats.metricType(stream, name);
stream.write(name).write("{cluster=\"").write(cluster).write("\",namespace=\"").write(namespace);
stream.write("\",remote_cluster=\"").write(remoteCluster).write("\"} ");
stream.write(value).write(' ').write(System.currentTimeMillis()).write('\n');
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java
index ff946b9..5968b8c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java
@@ -20,12 +20,10 @@ package org.apache.pulsar.broker.stats.prometheus;
import java.util.HashMap;
import java.util.Map;
-import java.util.Set;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerMBeanImpl;
import org.apache.bookkeeper.mledger.util.StatsBuckets;
import org.apache.pulsar.common.util.SimpleTextOutputStream;
-import org.eclipse.jetty.util.ConcurrentHashSet;
class TopicStats {
@@ -47,7 +45,6 @@ class TopicStats {
Map<String, AggregatedReplicationStats> replicationStats = new HashMap<>();
Map<String, AggregatedSubscriptionStats> subscriptionStats = new HashMap<>();
- private static Set<String> METRIC_TYPES = new ConcurrentHashSet<>();
public void reset() {
subscriptionsCount = 0;
@@ -71,9 +68,6 @@ class TopicStats {
static void printTopicStats(SimpleTextOutputStream stream, String cluster, String namespace, String topic,
TopicStats stats) {
-
- METRIC_TYPES.forEach(metric -> metricType(stream, metric));
-
metric(stream, cluster, namespace, topic, "pulsar_subscriptions_count", stats.subscriptionsCount);
metric(stream, cluster, namespace, topic, "pulsar_producers_count", stats.producersCount);
metric(stream, cluster, namespace, topic, "pulsar_consumers_count", stats.consumersCount);
@@ -140,11 +134,7 @@ class TopicStats {
private static void metric(SimpleTextOutputStream stream, String cluster, String namespace, String topic,
String name, double value) {
- if (!METRIC_TYPES.contains(name)) {
- metricType(stream, name);
- METRIC_TYPES.add(name);
- }
-
+ metricType(stream, name);
stream.write(name).write("{cluster=\"").write(cluster).write("\",namespace=\"").write(namespace)
.write("\",topic=\"").write(topic).write("\"} ");
stream.write(value).write(' ').write(System.currentTimeMillis()).write('\n');
@@ -152,11 +142,7 @@ class TopicStats {
private static void metric(SimpleTextOutputStream stream, String cluster, String namespace, String topic, String subscription,
String name, long value) {
- if (!METRIC_TYPES.contains(name)) {
- metricType(stream, name);
- METRIC_TYPES.add(name);
- }
-
+ metricType(stream, name);
stream.write(name).write("{cluster=\"").write(cluster).write("\",namespace=\"").write(namespace)
.write("\",topic=\"").write(topic).write("\",subscription=\"").write(subscription).write("\"} ");
stream.write(value).write(' ').write(System.currentTimeMillis()).write('\n');
@@ -164,11 +150,7 @@ class TopicStats {
private static void metric(SimpleTextOutputStream stream, String cluster, String namespace, String topic, String subscription,
String name, double value) {
- if (!METRIC_TYPES.contains(name)) {
- metricType(stream, name);
- METRIC_TYPES.add(name);
- }
-
+ metricType(stream, name);
stream.write(name).write("{cluster=\"").write(cluster).write("\",namespace=\"").write(namespace)
.write("\",topic=\"").write(topic).write("\",subscription=\"").write(subscription).write("\"} ");
stream.write(value).write(' ').write(System.currentTimeMillis()).write('\n');
@@ -176,11 +158,7 @@ class TopicStats {
private static void metric(SimpleTextOutputStream stream, String cluster, String namespace, String topic, String subscription,
String consumerName, long consumerId, String name, long value) {
- if (!METRIC_TYPES.contains(name)) {
- metricType(stream, name);
- METRIC_TYPES.add(name);
- }
-
+ metricType(stream, name);
stream.write(name).write("{cluster=\"").write(cluster).write("\", namespace=\"").write(namespace)
.write("\",topic=\"").write(topic).write("\",subscription=\"").write(subscription)
.write("\",consumer_name=\"").write(consumerName).write("\",consumer_id=\"").write(consumerId).write("\"} ");
@@ -189,11 +167,7 @@ class TopicStats {
private static void metric(SimpleTextOutputStream stream, String cluster, String namespace, String topic, String subscription,
String consumerName, long consumerId, String name, double value) {
- if (!METRIC_TYPES.contains(name)) {
- metricType(stream, name);
- METRIC_TYPES.add(name);
- }
-
+ metricType(stream, name);
stream.write(name).write("{cluster=\"").write(cluster).write("\",namespace=\"").write(namespace)
.write("\",topic=\"").write(topic).write("\",subscription=\"").write(subscription)
.write("\",consumer_name=\"").write(consumerName).write("\",consumer_id=\"").write(consumerId).write("\"} ");
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionsStatsGenerator.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionsStatsGenerator.java
index 907afaf..be7c88b 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionsStatsGenerator.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionsStatsGenerator.java
@@ -37,12 +37,8 @@ public class FunctionsStatsGenerator {
private static final Logger log = LoggerFactory.getLogger(FunctionsStatsGenerator.class);
- private static Set<String> METRIC_TYPES = new ConcurrentHashSet<>();
-
public static void generate(WorkerService workerService, String cluster, SimpleTextOutputStream out) {
if (workerService != null) {
- METRIC_TYPES.forEach(metric -> metricType(out, metric));
-
Map<String, FunctionRuntimeInfo> functionRuntimes
= workerService.getFunctionRuntimeManager().getFunctionRuntimeInfos();
@@ -98,11 +94,7 @@ public class FunctionsStatsGenerator {
private static void metric(SimpleTextOutputStream stream, String cluster, String namespace,
String functionName, String metricName, int instanceId, double value) {
- if (!METRIC_TYPES.contains(metricName)) {
- metricType(stream, metricName);
- METRIC_TYPES.add(metricName);
- }
-
+ metricType(stream, metricName);
stream.write(metricName).write("{cluster=\"").write(cluster).write("\",namespace=\"").write(namespace)
.write("\",name=\"").write(functionName).write("\",instanceId=\"").write(instanceId).write("\"} ");
stream.write(value).write(' ').write(System.currentTimeMillis()).write('\n');
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsMetricsResource.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsMetricsResource.java
index 0b87060..b94dfd8 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsMetricsResource.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsMetricsResource.java
@@ -53,7 +53,10 @@ public class FunctionsMetricsResource extends FunctionApiResource {
out.write(payload, arrayOffset, readableBytes);
out.flush();
};
- return Response.ok(streamOut).build();
+ return Response
+ .ok(streamOut)
+ .type(MediaType.TEXT_PLAIN_TYPE)
+ .build();
} finally {
buf.release();
}