From 8e55adc234ce6457ec656f6f4a647fcc85928a11 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Tue, 23 Apr 2024 10:55:24 +0800 Subject: [PATCH 1/7] add role. --- .../main/java/org/apache/pulsar/broker/service/Consumer.java | 1 + .../org/apache/pulsar/common/policies/data/ConsumerStats.java | 3 +++ .../pulsar/common/policies/data/stats/ConsumerStatsImpl.java | 3 +++ 3 files changed, 7 insertions(+) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java index 6b2028095e205..c9a7f5b224109 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java @@ -200,6 +200,7 @@ public Consumer(Subscription subscription, SubType subType, String topicName, lo stats = new ConsumerStatsImpl(); stats.setAddress(cnx.clientSourceAddressAndPort()); stats.consumerName = consumerName; + stats.role = appId; stats.setConnectedSince(DateFormatter.now()); stats.setClientVersion(cnx.getClientVersion()); stats.metadata = this.metadata; diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ConsumerStats.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ConsumerStats.java index 8c9a615d6d01c..bca7c7bf46f21 100644 --- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ConsumerStats.java +++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ConsumerStats.java @@ -25,6 +25,9 @@ * Consumer statistics. */ public interface ConsumerStats { + /** the role used for authorization. */ + String getRole(); + /** Total rate of messages delivered to the consumer (msg/s). */ double getMsgRateOut(); diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java index 548abdc9ada33..a4f6bd3b2ae6a 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java @@ -30,6 +30,9 @@ */ @Data public class ConsumerStatsImpl implements ConsumerStats { + /** the role used for authorization. */ + public String role; + /** Total rate of messages delivered to the consumer (msg/s). */ public double msgRateOut; From ae55a6dee2fee6844cf38b60113d8c5d5848698f Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Tue, 23 Apr 2024 12:04:28 +0800 Subject: [PATCH 2/7] add test. --- .../stats/AuthenticatedConsumerStatsTest.java | 151 ++++++++++++++++++ 1 file changed, 151 insertions(+) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/AuthenticatedConsumerStatsTest.java diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/AuthenticatedConsumerStatsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/AuthenticatedConsumerStatsTest.java new file mode 100644 index 0000000000000..55b3f86bcdb98 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/AuthenticatedConsumerStatsTest.java @@ -0,0 +1,151 @@ +package org.apache.pulsar.broker.stats; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.collect.Sets; +import io.jsonwebtoken.Jwts; +import io.jsonwebtoken.SignatureAlgorithm; +import org.apache.pulsar.broker.authentication.AuthenticationProviderToken; +import org.apache.pulsar.client.admin.PulsarAdminBuilder; +import org.apache.pulsar.client.api.AuthenticationFactory; +import org.apache.pulsar.client.api.ClientBuilder; +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.client.impl.auth.AuthenticationToken; +import org.apache.pulsar.common.policies.data.ConsumerStats; +import org.apache.pulsar.common.policies.data.TopicStats; +import org.apache.pulsar.common.util.ObjectMapperFactory; +import org.testng.Assert; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +import java.security.KeyPair; +import java.security.KeyPairGenerator; +import java.security.NoSuchAlgorithmException; +import java.security.PrivateKey; +import java.time.Duration; +import java.util.Base64; +import java.util.Date; +import java.util.HashSet; +import java.util.Iterator; +import java.util.Properties; +import java.util.Set; + +public class AuthenticatedConsumerStatsTest extends ConsumerStatsTest{ + private final String ADMIN_TOKEN; + private final String TOKEN_PUBLIC_KEY; + private final KeyPair kp; + + AuthenticatedConsumerStatsTest() throws NoSuchAlgorithmException { + KeyPairGenerator kpg = KeyPairGenerator.getInstance("RSA"); + kp = kpg.generateKeyPair(); + + byte[] encodedPublicKey = kp.getPublic().getEncoded(); + TOKEN_PUBLIC_KEY = "data:;base64," + Base64.getEncoder().encodeToString(encodedPublicKey); + ADMIN_TOKEN = generateToken(kp, "admin"); + } + + + private String generateToken(KeyPair kp, String subject) { + PrivateKey pkey = kp.getPrivate(); + long expMillis = System.currentTimeMillis() + Duration.ofHours(1).toMillis(); + Date exp = new Date(expMillis); + + return Jwts.builder() + .setSubject(subject) + .setExpiration(exp) + .signWith(pkey, SignatureAlgorithm.forSigningKey(pkey)) + .compact(); + } + + @Override + protected void customizeNewPulsarClientBuilder(ClientBuilder clientBuilder) { + clientBuilder.authentication(AuthenticationFactory.token(ADMIN_TOKEN)); + } + + @Override + protected void customizeNewPulsarAdminBuilder(PulsarAdminBuilder pulsarAdminBuilder) { + pulsarAdminBuilder.authentication(AuthenticationFactory.token(ADMIN_TOKEN)); + } + + @BeforeMethod + @Override + protected void setup() throws Exception { + conf.setAuthenticationEnabled(true); + conf.setAuthorizationEnabled(true); + + Set superUserRoles = new HashSet<>(); + superUserRoles.add("admin"); + conf.setSuperUserRoles(superUserRoles); + + Set providers = new HashSet<>(); + providers.add(AuthenticationProviderToken.class.getName()); + conf.setAuthenticationProviders(providers); + conf.setBrokerClientAuthenticationPlugin(AuthenticationToken.class.getName()); + conf.setBrokerClientAuthenticationParameters("token:" + ADMIN_TOKEN); + + conf.setClusterName("test"); + + // Set provider domain name + Properties properties = new Properties(); + properties.setProperty("tokenPublicKey", TOKEN_PUBLIC_KEY); + conf.setProperties(properties); + + super.internalSetup(); + super.producerBaseSetup(); + } + + @Test + public void testConsumerStatsOutput() throws Exception { + Set allowedFields = Sets.newHashSet( + "msgRateOut", + "msgThroughputOut", + "bytesOutCounter", + "msgOutCounter", + "messageAckRate", + "msgRateRedeliver", + "chunkedMessageRate", + "consumerName", + "availablePermits", + "unackedMessages", + "avgMessagesPerEntry", + "blockedConsumerOnUnackedMsgs", + "readPositionWhenJoining", + "lastAckedTime", + "lastAckedTimestamp", + "lastConsumedTime", + "lastConsumedTimestamp", + "lastConsumedFlowTimestamp", + "keyHashRanges", + "metadata", + "address", + "connectedSince", + "clientVersion", + "role"); + + final String topicName = "persistent://public/default/testConsumerStatsOutput"; + final String subName = "my-subscription"; + + Consumer consumer = pulsarClient.newConsumer() + .topic(topicName) + .subscriptionType(SubscriptionType.Shared) + .subscriptionName(subName) + .subscribe(); + + TopicStats stats = admin.topics().getStats(topicName); + ObjectMapper mapper = ObjectMapperFactory.create(); + ConsumerStats consumerStats = stats.getSubscriptions() + .get(subName).getConsumers().get(0); + Assert.assertTrue(consumerStats.getLastConsumedFlowTimestamp() > 0); + JsonNode node = mapper.readTree(mapper.writer().writeValueAsString(consumerStats)); + Iterator itr = node.fieldNames(); + while (itr.hasNext()) { + String field = itr.next(); + Assert.assertTrue(allowedFields.contains(field), field + " should not be exposed"); + } + // assert that role is exposed + Assert.assertEquals(consumerStats.getRole(), "admin"); + consumer.close(); + } + +} From e6a56c6005c09e5f2e7ee8ab49cd297ea6421b5a Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Tue, 23 Apr 2024 14:16:17 +0800 Subject: [PATCH 3/7] add license. --- .../stats/AuthenticatedConsumerStatsTest.java | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/AuthenticatedConsumerStatsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/AuthenticatedConsumerStatsTest.java index 55b3f86bcdb98..b516aa32c74cf 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/AuthenticatedConsumerStatsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/AuthenticatedConsumerStatsTest.java @@ -1,3 +1,21 @@ +/* + * 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.stats; import com.fasterxml.jackson.databind.JsonNode; From 854a0560eb77174d7207dafc344b242ae6f92b9f Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Mon, 13 May 2024 18:04:33 +0800 Subject: [PATCH 4/7] rename role to appId. --- .../main/java/org/apache/pulsar/broker/service/Consumer.java | 2 +- .../pulsar/broker/stats/AuthenticatedConsumerStatsTest.java | 4 ++-- .../org/apache/pulsar/common/policies/data/ConsumerStats.java | 4 ++-- .../pulsar/common/policies/data/stats/ConsumerStatsImpl.java | 4 ++-- 4 files changed, 7 insertions(+), 7 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java index 8b55bd9eaf2d8..ef449463a54d9 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java @@ -200,7 +200,7 @@ public Consumer(Subscription subscription, SubType subType, String topicName, lo stats = new ConsumerStatsImpl(); stats.setAddress(cnx.clientSourceAddressAndPort()); stats.consumerName = consumerName; - stats.role = appId; + stats.appId = appId; stats.setConnectedSince(DateFormatter.now()); stats.setClientVersion(cnx.getClientVersion()); stats.metadata = this.metadata; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/AuthenticatedConsumerStatsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/AuthenticatedConsumerStatsTest.java index b516aa32c74cf..e8cadb72e1e04 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/AuthenticatedConsumerStatsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/AuthenticatedConsumerStatsTest.java @@ -139,7 +139,7 @@ public void testConsumerStatsOutput() throws Exception { "address", "connectedSince", "clientVersion", - "role"); + "appId"); final String topicName = "persistent://public/default/testConsumerStatsOutput"; final String subName = "my-subscription"; @@ -162,7 +162,7 @@ public void testConsumerStatsOutput() throws Exception { Assert.assertTrue(allowedFields.contains(field), field + " should not be exposed"); } // assert that role is exposed - Assert.assertEquals(consumerStats.getRole(), "admin"); + Assert.assertEquals(consumerStats.getAppId(), "admin"); consumer.close(); } diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ConsumerStats.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ConsumerStats.java index bca7c7bf46f21..d2d3600df96ed 100644 --- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ConsumerStats.java +++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ConsumerStats.java @@ -25,8 +25,8 @@ * Consumer statistics. */ public interface ConsumerStats { - /** the role used for authorization. */ - String getRole(); + /** the app id. */ + String getAppId(); /** Total rate of messages delivered to the consumer (msg/s). */ double getMsgRateOut(); diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java index a4f6bd3b2ae6a..de36b330b7f1a 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java @@ -30,8 +30,8 @@ */ @Data public class ConsumerStatsImpl implements ConsumerStats { - /** the role used for authorization. */ - public String role; + /** the app id. */ + public String appId; /** Total rate of messages delivered to the consumer (msg/s). */ public double msgRateOut; From 677ed2f733c499a72513c758635a88a4e1cb265c Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Tue, 14 May 2024 19:54:18 +0800 Subject: [PATCH 5/7] fix conflict. --- .../pulsar/broker/service/Consumer.java | 34 +++++++++++++++++-- 1 file changed, 32 insertions(+), 2 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java index ef449463a54d9..c9f417c4bc4f7 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java @@ -25,6 +25,7 @@ import com.google.common.util.concurrent.AtomicDouble; import io.netty.util.concurrent.Future; import io.netty.util.concurrent.Promise; +import java.time.Instant; import java.util.ArrayList; import java.util.BitSet; import java.util.Collections; @@ -90,7 +91,9 @@ public class Consumer { private final Rate msgOut; private final Rate msgRedeliver; private final LongAdder msgOutCounter; + private final LongAdder msgRedeliverCounter; private final LongAdder bytesOutCounter; + private final LongAdder messageAckCounter; private final Rate messageAckRate; private volatile long lastConsumedTimestamp; @@ -152,6 +155,9 @@ public class Consumer { @Getter private final SchemaType schemaType; + @Getter + private final Instant connectedSince = Instant.now(); + public Consumer(Subscription subscription, SubType subType, String topicName, long consumerId, int priorityLevel, String consumerName, boolean isDurable, TransportCnx cnx, String appId, @@ -182,8 +188,10 @@ public Consumer(Subscription subscription, SubType subType, String topicName, lo this.msgOut = new Rate(); this.chunkedMessageRate = new Rate(); this.msgRedeliver = new Rate(); + this.msgRedeliverCounter = new LongAdder(); this.bytesOutCounter = new LongAdder(); this.msgOutCounter = new LongAdder(); + this.messageAckCounter = new LongAdder(); this.messageAckRate = new Rate(); this.appId = appId; @@ -201,7 +209,7 @@ public Consumer(Subscription subscription, SubType subType, String topicName, lo stats.setAddress(cnx.clientSourceAddressAndPort()); stats.consumerName = consumerName; stats.appId = appId; - stats.setConnectedSince(DateFormatter.now()); + stats.setConnectedSince(DateFormatter.format(connectedSince)); stats.setClientVersion(cnx.getClientVersion()); stats.metadata = this.metadata; @@ -239,8 +247,10 @@ public Consumer(Subscription subscription, SubType subType, String topicName, lo this.consumerName = consumerName; this.msgOut = null; this.msgRedeliver = null; + this.msgRedeliverCounter = null; this.msgOutCounter = null; this.bytesOutCounter = null; + this.messageAckCounter = null; this.messageAckRate = null; this.pendingAcks = null; this.stats = null; @@ -503,6 +513,7 @@ public CompletableFuture messageAcked(CommandAck ack) { return future .thenApply(v -> { this.messageAckRate.recordEvent(v); + this.messageAckCounter.add(v); return null; }); } @@ -923,6 +934,14 @@ public long getBytesOutCounter() { return bytesOutCounter.longValue(); } + public long getMessageAckCounter() { + return messageAckCounter.sum(); + } + + public long getMessageRedeliverCounter() { + return msgRedeliverCounter.sum(); + } + public int getUnackedMessages() { return unackedMessages; } @@ -935,7 +954,7 @@ public KeySharedMeta getKeySharedMeta() { public String toString() { if (subscription != null && cnx != null) { return MoreObjects.toStringHelper(this).add("subscription", subscription).add("consumerId", consumerId) - .add("consumerName", consumerName).add("address", this.cnx.clientAddress()).toString(); + .add("consumerName", consumerName).add("address", this.cnx.toString()).toString(); } else { return MoreObjects.toStringHelper(this).add("consumerId", consumerId) .add("consumerName", consumerName).toString(); @@ -1060,6 +1079,8 @@ public void redeliverUnacknowledgedMessages(long consumerEpoch) { } msgRedeliver.recordMultipleEvents(totalRedeliveryMessages.intValue(), totalRedeliveryMessages.intValue()); + msgRedeliverCounter.add(totalRedeliveryMessages.intValue()); + subscription.redeliverUnacknowledgedMessages(this, pendingPositions); } else { subscription.redeliverUnacknowledgedMessages(this, consumerEpoch); @@ -1092,6 +1113,7 @@ public void redeliverUnacknowledgedMessages(List messageIds) { subscription.redeliverUnacknowledgedMessages(this, pendingPositions); msgRedeliver.recordMultipleEvents(totalRedeliveryMessages, totalRedeliveryMessages); + msgRedeliverCounter.add(totalRedeliveryMessages); int numberOfBlockedPermits = PERMITS_RECEIVED_WHILE_CONSUMER_BLOCKED_UPDATER.getAndSet(this, 0); @@ -1154,6 +1176,14 @@ public String getClientAddress() { return clientAddress; } + public String getClientAddressAndPort() { + return cnx.clientSourceAddressAndPort(); + } + + public String getClientVersion() { + return cnx.getClientVersion(); + } + public MessageId getStartMessageId() { return startMessageId; } From 92d7a880f4dbf6cb797dcfddf2dd5895e5446f09 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Wed, 15 May 2024 09:58:51 +0800 Subject: [PATCH 6/7] append vote thread. --- pip/pip-347.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pip/pip-347.md b/pip/pip-347.md index 5326fed353374..a5d5d76ae1700 100644 --- a/pip/pip-347.md +++ b/pip/pip-347.md @@ -34,4 +34,4 @@ Fully compatible. Updated afterwards --> * Mailing List discussion thread: https://lists.apache.org/thread/p9y9r8pb7ygk8f0jd121c1121phvzd09 -* Mailing List voting thread: +* Mailing List voting thread: https://lists.apache.org/thread/sfv0vq498dnjx6k6zdrnn0cw8f22tz05 From 50b726568c518a5df0a8a4e1ce8eabecd7d39d77 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Wed, 15 May 2024 16:20:30 +0800 Subject: [PATCH 7/7] fix test code. --- .../java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java index 512a5cfcab661..024d8582fa213 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java @@ -195,7 +195,7 @@ public void testAckStatsOnPartitionedTopicForExclusiveSubscription() throws Puls @Test public void testUpdateStatsForActiveConsumerAndSubscription() throws Exception { - final String topicName = "persistent://prop/use/ns-abc/testUpdateStatsForActiveConsumerAndSubscription"; + final String topicName = "persistent://public/default/testUpdateStatsForActiveConsumerAndSubscription"; pulsarClient.newConsumer() .topic(topicName) .subscriptionType(SubscriptionType.Shared)