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/04/14 00:10:09 UTC
[incubator-pulsar] branch master updated: Fix bug that WebSocket
proxy returns empty metrics (#1567)
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/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new aa9e306 Fix bug that WebSocket proxy returns empty metrics (#1567)
aa9e306 is described below
commit aa9e3065931bdad51cf83daeb4efcee96b761748
Author: massakam <ma...@yahoo-corp.jp>
AuthorDate: Sat Apr 14 09:10:05 2018 +0900
Fix bug that WebSocket proxy returns empty metrics (#1567)
---
.../websocket/proxy/ProxyPublishConsumeTest.java | 12 +++++--
.../apache/pulsar/websocket/stats/ProxyStats.java | 40 ++++++++++++++++++----
2 files changed, 42 insertions(+), 10 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java
index 1e5409c..23895fc 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java
@@ -446,9 +446,15 @@ public class ProxyPublishConsumeTest extends ProducerConsumerBase {
Response response = (Response) invocationBuilder.get();
String responseStr = response.readEntity(String.class);
final Gson gson = new Gson();
- System.err.println("REQ: " + statUrl);
- System.err.println("RESPONSE: " + responseStr);
- final List<Metrics> data = gson.fromJson(responseStr, new TypeToken<List<Metrics>>() {
+ List<Metrics> data = gson.fromJson(responseStr, new TypeToken<List<Metrics>>() {
+ }.getType());
+ Assert.assertFalse(data.isEmpty());
+ // re-generate metrics
+ service.getProxyStats().generate();
+ invocationBuilder = webTarget.request(MediaType.APPLICATION_JSON);
+ response = (Response) invocationBuilder.get();
+ responseStr = response.readEntity(String.class);
+ data = gson.fromJson(responseStr, new TypeToken<List<Metrics>>() {
}.getType());
Assert.assertFalse(data.isEmpty());
}
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/stats/ProxyStats.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/stats/ProxyStats.java
index 76f1bb2..b1816c8 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/stats/ProxyStats.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/stats/ProxyStats.java
@@ -32,6 +32,9 @@ import org.apache.pulsar.websocket.WebSocketService;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
/**
* It periodically generates stats metrics of proxy service,
*
@@ -41,7 +44,6 @@ public class ProxyStats {
private final WebSocketService service;
private final JvmMetrics jvmMetrics;
private ConcurrentOpenHashMap<String, ProxyNamespaceStats> topicStats;
- private List<Metrics> tempMetricsCollection;
private List<Metrics> metricsCollection;
public ProxyStats(WebSocketService service) {
@@ -49,7 +51,6 @@ public class ProxyStats {
this.service = service;
this.jvmMetrics = new JvmMetrics(service);
this.topicStats = new ConcurrentOpenHashMap<>();
- this.tempMetricsCollection = Lists.newArrayList();
this.metricsCollection = Lists.newArrayList();
// schedule stat generation task every 1 minute
service.getExecutor().scheduleAtFixedRate(() -> generate(), 120, 60, TimeUnit.SECONDS);
@@ -59,25 +60,32 @@ public class ProxyStats {
* generates stats-metrics of proxy service and updates metricsCollection cache with latest stats.
*/
public synchronized void generate() {
+ if (log.isDebugEnabled()) {
+ log.debug("Start generating proxy metrics");
+ }
topicStats.clear();
service.getProducers().forEach((topic, handlers) -> {
+ if (log.isDebugEnabled()) {
+ log.debug("Collect stats from {} producer handlers for topic {}", handlers.size(), topic);
+ }
+
final String namespaceName = TopicName.get(topic).getNamespace();
ProxyNamespaceStats nsStat = topicStats.computeIfAbsent(namespaceName, ns -> new ProxyNamespaceStats());
handlers.forEach(handler -> {
nsStat.numberOfMsgPublished += handler.getAndResetNumMsgsSent();
nsStat.numberOfBytesPublished += handler.getAndResetNumBytesSent();
nsStat.numberOfPublishFailure += handler.getAndResetNumMsgsFailed();
- if (nsStat.publishMsgLatency == null) {
- nsStat.publishMsgLatency = new StatsBuckets(ENTRY_LATENCY_BUCKETS_USEC);
- }
handler.getPublishLatencyStatsUSec().refresh();
nsStat.publishMsgLatency.addAll(handler.getPublishLatencyStatsUSec());
- System.out.println(nsStat.publishMsgLatency);
});
});
service.getConsumers().forEach((topic, handlers) -> {
+ if (log.isDebugEnabled()) {
+ log.debug("Collect stats from {} consumer handlers for topic {}", handlers.size(), topic);
+ }
+
final String namespaceName = TopicName.get(topic).getNamespace();
ProxyNamespaceStats nsStat = topicStats.computeIfAbsent(namespaceName, ns -> new ProxyNamespaceStats());
handlers.forEach(handler -> {
@@ -87,16 +95,28 @@ public class ProxyStats {
});
});
- tempMetricsCollection.clear();
+ List<Metrics> tempMetricsCollection = Lists.newArrayList();
topicStats.forEach((namespace, stats) -> {
+ if (log.isDebugEnabled()) {
+ log.debug("Add ns-stats of namespace {} to metrics", namespace);
+ }
tempMetricsCollection.add(stats.add(namespace));
});
+
// add jvm-metrics
+ if (log.isDebugEnabled()) {
+ log.debug("Add jvm-stats to metrics");
+ }
tempMetricsCollection.add(jvmMetrics.generate());
+
// swap tempmetrics to stat-metrics
List<Metrics> tempRef = metricsCollection;
metricsCollection = tempMetricsCollection;
tempRef.clear();
+
+ if (log.isDebugEnabled()) {
+ log.debug("Complete generating proxy metrics");
+ }
}
public synchronized List<Metrics> getMetrics() {
@@ -114,6 +134,10 @@ public class ProxyStats {
public long numberOfBytesDelivered;
public long numberOfMsgsAcked;
+ public ProxyNamespaceStats() {
+ this.publishMsgLatency = new StatsBuckets(ENTRY_LATENCY_BUCKETS_USEC);
+ }
+
public Metrics add(String namespace) {
publishMsgLatency.refresh();
@@ -137,4 +161,6 @@ public class ProxyStats {
}
}
+ private static final Logger log = LoggerFactory.getLogger(ProxyStats.class);
+
}
--
To stop receiving notification emails like this one, please contact
mmerli@apache.org.