You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "hanmz (via GitHub)" <gi...@apache.org> on 2024/04/16 13:00:57 UTC
[PR] [improve][broker] Optimize topic production msgRate stats. [pulsar]
hanmz opened a new pull request, #22518:
URL: https://github.com/apache/pulsar/pull/22518
### Motivation
In a production environment, we often encounter a scenario: send a message to create a producer, and then close it after use. At this time, we will see that the `msgRateIn` and `msgThroughputIn` in topics stats are always 0. This is because the msgRateIn and msgThroughputIn of the topic are calculated from the online producer.
(Here I only optimized the production stats, and the consumption stats also have similar problems.)
### Modifications
1. We add `rateIn` statistics in topic to avoid relying on producers being online.
2. Here we also add an optional configuration to be compatible with the previous logic.
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Optimize topic production rate stats. [pulsar]
Posted by "hanmz (via GitHub)" <gi...@apache.org>.
hanmz commented on code in PR #22518:
URL: https://github.com/apache/pulsar/pull/22518#discussion_r1568325708
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -2391,6 +2395,12 @@ public CompletableFuture<? extends TopicStatsImpl> asyncGetStats(GetStatsOptions
}
});
+ // Replace producer stats with topic-level stats
Review Comment:
all right
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Optimize topic production rate stats. [pulsar]
Posted by "hanmz (via GitHub)" <gi...@apache.org>.
hanmz commented on PR #22518:
URL: https://github.com/apache/pulsar/pull/22518#issuecomment-2060677786
> Plz add unit tests to cover this logic.
I have added unit tests, PLAT~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Optimize topic production rate stats. [pulsar]
Posted by "hanmz (via GitHub)" <gi...@apache.org>.
hanmz closed pull request #22518: [improve][broker] Optimize topic production rate stats.
URL: https://github.com/apache/pulsar/pull/22518
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Optimize topic production rate stats. [pulsar]
Posted by "hanmz (via GitHub)" <gi...@apache.org>.
hanmz commented on code in PR #22518:
URL: https://github.com/apache/pulsar/pull/22518#discussion_r1568323619
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java:
##########
@@ -885,6 +887,13 @@ public void recordAddLatency(long latency, TimeUnit unit) {
PUBLISH_LATENCY.observe(latency, unit);
}
+ @Override
+ public void recordRateIn(long events, long totalValue) {
Review Comment:
OK
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Optimize topic production rate stats. [pulsar]
Posted by "hanmz (via GitHub)" <gi...@apache.org>.
hanmz commented on code in PR #22518:
URL: https://github.com/apache/pulsar/pull/22518#discussion_r1568324951
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java:
##########
@@ -885,6 +887,13 @@ public void recordAddLatency(long latency, TimeUnit unit) {
PUBLISH_LATENCY.observe(latency, unit);
}
+ @Override
+ public void recordRateIn(long events, long totalValue) {
+ if (brokerService.pulsar().getConfig().isEnableReplaceProducerStatsWithTopicStats()) {
Review Comment:
alright
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Optimize topic production rate stats. [pulsar]
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #22518:
URL: https://github.com/apache/pulsar/pull/22518#discussion_r1569087887
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java:
##########
@@ -1899,4 +1900,106 @@ public void testUpdatePropertiesOnNonDurableSub() throws Exception {
assertEquals(cursor.getCursorProperties().size(), 1);
assertEquals(cursor.getCursorProperties().get("foo"), "bar");
}
+
+ @Test
+ public void testPrecomputeProducerStatsInTopicStatsForPersistentTopic() throws Exception {
+ final int numberOfMessages = 10;
+ final String testLocalTopicName = "testPrecomputeProducerStatsInTopicStatsForPersistentTopic";
+ final String topicName = "persistent://" + testTenant + "/" + testNamespace + "/" + testLocalTopicName;
+ admin.topics().createNonPartitionedTopic(topicName);
+
+ // 1) Use producer stats to compute msgRateIn and msgThroughputIn for persistentTopic
+ pulsar.getConfiguration().setPrecomputeProducerStatsInTopicStats(false);
+
+ // produce numberOfMessages message to pulsar
+ ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer().topic(topicName).create();
+ for (int i = 0; i < numberOfMessages; i++) {
+ log.info("Produce messages: " + producer.send(new byte[10]).toString());
+ }
+ producer.close();
+
+ // trigger update rates
+ pulsar.getBrokerService().updateRates();
+
+ AsyncResponse response = mock(AsyncResponse.class);
+ persistentTopics.getStats(response, testTenant, testNamespace, testLocalTopicName, true, true, false, false, false, false);
+ ArgumentCaptor<TopicStats> statCaptor = ArgumentCaptor.forClass(TopicStats.class);
+ verify(response, timeout(5000).times(1)).resume(statCaptor.capture());
+ TopicStats topicStats = statCaptor.getValue();
+ assertEquals(topicStats.getMsgRateIn(), 0);
+ assertEquals(topicStats.getMsgThroughputIn(), 0);
+
Review Comment:
Plz cover other publisher stats: bytesInCounter, msgInCounter, and averageMsgSize
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java:
##########
@@ -1899,4 +1900,106 @@ public void testUpdatePropertiesOnNonDurableSub() throws Exception {
assertEquals(cursor.getCursorProperties().size(), 1);
assertEquals(cursor.getCursorProperties().get("foo"), "bar");
}
+
+ @Test
+ public void testPrecomputeProducerStatsInTopicStatsForPersistentTopic() throws Exception {
+ final int numberOfMessages = 10;
+ final String testLocalTopicName = "testPrecomputeProducerStatsInTopicStatsForPersistentTopic";
+ final String topicName = "persistent://" + testTenant + "/" + testNamespace + "/" + testLocalTopicName;
+ admin.topics().createNonPartitionedTopic(topicName);
+
+ // 1) Use producer stats to compute msgRateIn and msgThroughputIn for persistentTopic
+ pulsar.getConfiguration().setPrecomputeProducerStatsInTopicStats(false);
+
+ // produce numberOfMessages message to pulsar
+ ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer().topic(topicName).create();
+ for (int i = 0; i < numberOfMessages; i++) {
+ log.info("Produce messages: " + producer.send(new byte[10]).toString());
+ }
+ producer.close();
+
+ // trigger update rates
+ pulsar.getBrokerService().updateRates();
+
+ AsyncResponse response = mock(AsyncResponse.class);
+ persistentTopics.getStats(response, testTenant, testNamespace, testLocalTopicName, true, true, false, false, false, false);
+ ArgumentCaptor<TopicStats> statCaptor = ArgumentCaptor.forClass(TopicStats.class);
+ verify(response, timeout(5000).times(1)).resume(statCaptor.capture());
+ TopicStats topicStats = statCaptor.getValue();
+ assertEquals(topicStats.getMsgRateIn(), 0);
+ assertEquals(topicStats.getMsgThroughputIn(), 0);
+
+
+ // 2) Use topic stats to compute msgRateIn and msgThroughputIn for persistentTopic
+ pulsar.getConfiguration().setPrecomputeProducerStatsInTopicStats(true);
+
+ // produce numberOfMessages message to pulsar
+ producer = (ProducerImpl<byte[]>) pulsarClient.newProducer().topic(topicName).create();
+ for (int i = 0; i < numberOfMessages; i++) {
+ log.info("Produce messages: " + producer.send(new byte[10]).toString());
+ }
+ producer.close();
+
+ // trigger update rates
+ pulsar.getBrokerService().updateRates();
+
+ response = mock(AsyncResponse.class);
+ persistentTopics.getStats(response, testTenant, testNamespace, testLocalTopicName, true, true, false, false, false, false);
+ statCaptor = ArgumentCaptor.forClass(TopicStats.class);
+ verify(response, timeout(5000).times(1)).resume(statCaptor.capture());
+ topicStats = statCaptor.getValue();
+ Assert.assertTrue(topicStats.getMsgRateIn() > 0);
+ Assert.assertTrue(topicStats.getMsgThroughputIn() > 0);
+ }
Review Comment:
can you also confirm if `/metrics` API from BrokerStatsBase correctly returns?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Optimize topic production rate stats. [pulsar]
Posted by "hanmz (via GitHub)" <gi...@apache.org>.
hanmz commented on PR #22518:
URL: https://github.com/apache/pulsar/pull/22518#issuecomment-2072064765
> @hanmz This is a long-standing issue and the plan was to address as part of the overall changes of [PIP-264](https://github.com/apache/pulsar/blob/master/pip/pip-264.md). I wouldn't introduce a fix for just 1 metric with one dedicated config option when these metrics are going away soon. @dragosvictor has already started converting the metrics.
If the community will solve this problem in the near future, this PR will not be introduced.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Optimize topic production rate stats. [pulsar]
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #22518:
URL: https://github.com/apache/pulsar/pull/22518#discussion_r1567728500
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -2111,6 +2111,10 @@ public void updateRates(NamespaceStats nsStats, NamespaceBundleStats bundleStats
bundleStats.producerCount += producers.size();
topicStatsStream.startObject(topic);
+ if (brokerService.pulsar().getConfig().isEnableReplaceProducerStatsWithTopicStats()) {
+ rateIn.calculateRate();
Review Comment:
It seems like we don't skip/update topicStatsHelper.aggMsgRateIn and aggMsgThroughputIn here.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java:
##########
@@ -157,6 +158,7 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener<TopicP
protected volatile Pair<String, List<EntryFilter>> entryFilters;
protected volatile boolean transferring = false;
private volatile List<PublishRateLimiter> activeRateLimiters;
+ protected Rate rateIn = new Rate();
Review Comment:
can we rename it to `protected final Rate msgRateIn`?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java:
##########
@@ -885,6 +887,13 @@ public void recordAddLatency(long latency, TimeUnit unit) {
PUBLISH_LATENCY.observe(latency, unit);
}
+ @Override
+ public void recordRateIn(long events, long totalValue) {
+ if (brokerService.pulsar().getConfig().isEnableReplaceProducerStatsWithTopicStats()) {
Review Comment:
I think we should check this flag before calling this func.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java:
##########
@@ -885,6 +887,13 @@ public void recordAddLatency(long latency, TimeUnit unit) {
PUBLISH_LATENCY.observe(latency, unit);
}
+ @Override
+ public void recordRateIn(long events, long totalValue) {
Review Comment:
recordMsgRateIn
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java:
##########
@@ -3449,6 +3449,13 @@ The max allowed delay for delayed delivery (in milliseconds). If the broker rece
)
private Set<String> additionalServlets = new TreeSet<>();
+ @FieldContext(
+ category = CATEGORY_SERVER,
+ dynamic = true,
+ doc = "Enable or disable replace producer stats with topic stats when calculating topic production rate"
+ )
+ private boolean enableReplaceProducerStatsWithTopicStats = false;
Review Comment:
`precomputeProducerStatsInTopicStats` might be the better name here.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -2391,6 +2395,12 @@ public CompletableFuture<? extends TopicStatsImpl> asyncGetStats(GetStatsOptions
}
});
+ // Replace producer stats with topic-level stats
Review Comment:
If this flag is enabled, I think we can skip the above stats.msgRateIn and stats.msgThroughputIn computes to optimize.
```java
producers.values().forEach(producer -> {
PublisherStatsImpl publisherStats = producer.getStats();
if (!isEnableReplaceProducerStatsWithTopicStats ){
stats.msgRateIn += publisherStats.msgRateIn;
stats.msgThroughputIn += publisherStats.msgThroughputIn;
}
if (producer.isRemote()) {
remotePublishersStats.put(producer.getRemoteCluster(), publisherStats);
}
if (!getStatsOptions.isExcludePublishers()){
stats.addPublisher(publisherStats);
}
});
if (isEnableReplaceProducerStatsWithTopicStats ){
stats.msgRateIn = rateIn.getRate();
stats.msgThroughputIn = rateIn.getValueRate();
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Optimize topic production rate stats. [pulsar]
Posted by "hanmz (via GitHub)" <gi...@apache.org>.
hanmz commented on code in PR #22518:
URL: https://github.com/apache/pulsar/pull/22518#discussion_r1568317043
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java:
##########
@@ -3449,6 +3449,13 @@ The max allowed delay for delayed delivery (in milliseconds). If the broker rece
)
private Set<String> additionalServlets = new TreeSet<>();
+ @FieldContext(
+ category = CATEGORY_SERVER,
+ dynamic = true,
+ doc = "Enable or disable replace producer stats with topic stats when calculating topic production rate"
+ )
+ private boolean enableReplaceProducerStatsWithTopicStats = false;
Review Comment:
This is a good suggestion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Optimize topic production rate stats. [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on PR #22518:
URL: https://github.com/apache/pulsar/pull/22518#issuecomment-2061570161
@hanmz This is a long-standing issue and the plan was to address as part of the overall changes of [PIP-264](https://github.com/apache/pulsar/blob/master/pip/pip-264.md). I wouldn't introduce a fix for just 1 metric with one dedicated config option when these metrics are going away soon. @dragosvictor has already started converting the metrics.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org