From 76d35e1484155c84d9476766b15594b6e8483911 Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Thu, 16 May 2024 12:55:06 -0700 Subject: [PATCH 01/21] Add producer messaging counters --- .../pulsar/broker/service/Producer.java | 6 ++++ .../data/NonPersistentPublisherStats.java | 3 ++ .../common/policies/data/PublisherStats.java | 7 +++++ .../NonPersistentPublisherStatsImpl.java | 13 ++++++++ .../data/stats/PublisherStatsImpl.java | 31 +++++++++++++++++++ 5 files changed, 60 insertions(+) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java index c10e33818ed3a..90b852ee73d7b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java @@ -74,9 +74,12 @@ public class Producer { private final long producerId; private final String appId; private final BrokerInterceptor brokerInterceptor; + @Deprecated private Rate msgIn; + @Deprecated private Rate chunkedMessageRate; // it records msg-drop rate only for non-persistent topic + @Deprecated private final Rate msgDrop; private volatile long pendingPublishAcks = 0; @@ -374,6 +377,7 @@ private static final class MessagePublishContext implements PublishContext, Runn private long sequenceId; private long ledgerId; private long entryId; + @Deprecated private Rate rateIn; private int msgSize; private long batchSize; @@ -542,12 +546,14 @@ public void run() { // stats rateIn.recordMultipleEvents(batchSize, msgSize); + producer.stats.recordMsgIn(batchSize, msgSize); producer.topic.recordAddLatency(System.nanoTime() - startTimeNs, TimeUnit.NANOSECONDS); producer.cnx.getCommandSender().sendSendReceiptResponse(producer.producerId, sequenceId, highestSequenceId, ledgerId, entryId); producer.cnx.completedSendOperation(producer.isNonPersistentTopic, msgSize); if (this.chunked) { producer.chunkedMessageRate.recordEvent(); + producer.stats.recordChunkedMsgIn(1); } producer.publishOperationCompleted(); if (producer.brokerInterceptor != null) { diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/NonPersistentPublisherStats.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/NonPersistentPublisherStats.java index fec645c2e258d..ce82724cf4c0b 100644 --- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/NonPersistentPublisherStats.java +++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/NonPersistentPublisherStats.java @@ -27,4 +27,7 @@ public interface NonPersistentPublisherStats extends PublisherStats { * messages per connection. **/ double getMsgDropRate(); + + void recordMsgDrop(long numMessages); + long getMsgDropCount(); } \ No newline at end of file diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/PublisherStats.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/PublisherStats.java index 78af224d28c02..888d4642ea315 100644 --- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/PublisherStats.java +++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/PublisherStats.java @@ -60,4 +60,11 @@ public interface PublisherStats { /** Metadata (key/value strings) associated with this publisher. */ Map getMetadata(); + + void recordMsgIn(long messageCount, long byteCount); + long getMsgInCounter(); + long getBytesInCounter(); + + void recordChunkedMsgIn(long messageCount); + long getChunkedMsgIn(); } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/NonPersistentPublisherStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/NonPersistentPublisherStatsImpl.java index adf3f92ae71fc..66e94ace46d42 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/NonPersistentPublisherStatsImpl.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/NonPersistentPublisherStatsImpl.java @@ -20,6 +20,7 @@ import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.util.Objects; +import java.util.concurrent.atomic.LongAdder; import lombok.Getter; import org.apache.pulsar.common.policies.data.NonPersistentPublisherStats; @@ -35,10 +36,22 @@ public class NonPersistentPublisherStatsImpl extends PublisherStatsImpl implemen @Getter public double msgDropRate; + private final LongAdder msgDropCount = new LongAdder(); + public NonPersistentPublisherStatsImpl add(NonPersistentPublisherStatsImpl stats) { Objects.requireNonNull(stats); super.add(stats); this.msgDropRate += stats.msgDropRate; return this; } + + @Override + public void recordMsgDrop(long numMessages) { + msgDropCount.add(numMessages); + } + + @Override + public long getMsgDropCount() { + return msgDropCount.sum(); + } } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java index 304361bb2daec..db3a3a8469fc9 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java @@ -20,6 +20,7 @@ import com.fasterxml.jackson.annotation.JsonIgnore; import java.util.Map; +import java.util.concurrent.atomic.LongAdder; import lombok.Data; import org.apache.pulsar.client.api.ProducerAccessMode; import org.apache.pulsar.common.policies.data.PublisherStats; @@ -34,6 +35,10 @@ public class PublisherStatsImpl implements PublisherStats { public ProducerAccessMode accessMode; + private final LongAdder msgInCounter = new LongAdder(); + private final LongAdder bytesInCounter = new LongAdder(); + private final LongAdder chunkedMessageCounter = new LongAdder(); + /** Total rate of messages published by this publisher (msg/s). */ public double msgRateIn; @@ -107,4 +112,30 @@ public String getClientVersion() { public void setClientVersion(String clientVersion) { this.clientVersion = clientVersion; } + + @Override + public void recordMsgIn(long messageCount, long byteCount) { + msgInCounter.add(messageCount); + bytesInCounter.add(byteCount); + } + + @Override + public long getMsgInCounter() { + return msgInCounter.sum(); + } + + @Override + public long getBytesInCounter() { + return bytesInCounter.sum(); + } + + @Override + public void recordChunkedMsgIn(long messageCount) { + chunkedMessageCounter.add(messageCount); + } + + @Override + public long getChunkedMsgIn() { + return chunkedMessageCounter.sum(); + } } From c2251957e78990601dc83292f44819963fcd7da8 Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Mon, 20 May 2024 12:36:41 -0700 Subject: [PATCH 02/21] Add producer attributes --- .../OpenTelemetryAttributes.java | 25 +++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/pulsar-opentelemetry/src/main/java/org/apache/pulsar/opentelemetry/OpenTelemetryAttributes.java b/pulsar-opentelemetry/src/main/java/org/apache/pulsar/opentelemetry/OpenTelemetryAttributes.java index 4f898b382e633..2a584fef4a543 100644 --- a/pulsar-opentelemetry/src/main/java/org/apache/pulsar/opentelemetry/OpenTelemetryAttributes.java +++ b/pulsar-opentelemetry/src/main/java/org/apache/pulsar/opentelemetry/OpenTelemetryAttributes.java @@ -91,6 +91,31 @@ public interface OpenTelemetryAttributes { */ AttributeKey PULSAR_CONSUMER_CONNECTED_SINCE = AttributeKey.longKey("pulsar.consumer.connected_since"); + /** + * The name of the Pulsar producer. + */ + AttributeKey PULSAR_PRODUCER_NAME = AttributeKey.stringKey("pulsar.producer.name"); + + /** + * The ID of the Pulsar producer. + */ + AttributeKey PULSAR_PRODUCER_ID = AttributeKey.longKey("pulsar.producer.id"); + + /** + * The access mode of the Pulsar producer. + */ + AttributeKey PULSAR_PRODUCER_ACCESS_MODE = AttributeKey.stringKey("pulsar.producer.access_mode"); + + /** + * The producer metadata properties, as a list of "key:value" pairs. + */ + AttributeKey> PULSAR_PRODUCER_METADATA = AttributeKey.stringArrayKey("pulsar.producer.metadata"); + + /** + * The UTC timestamp of the Pulsar producer creation. + */ + AttributeKey PULSAR_PRODUCER_CONNECTED_SINCE = AttributeKey.longKey("pulsar.producer.connected_since"); + /** * The address of the Pulsar client. */ From d5abd2d3c675bbb87bbeb61990c353d68657f60b Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Mon, 20 May 2024 12:43:04 -0700 Subject: [PATCH 03/21] Add draft OpenTelemetryProducerStats --- .../stats/OpenTelemetryProducerStats.java | 122 ++++++++++++++++++ 1 file changed, 122 insertions(+) create mode 100644 pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java new file mode 100644 index 0000000000000..987898338d10f --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java @@ -0,0 +1,122 @@ +/* + * 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 io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.BatchCallback; +import io.opentelemetry.api.metrics.ObservableLongMeasurement; +import java.util.Optional; +import org.apache.pulsar.broker.PulsarService; +import org.apache.pulsar.broker.service.Producer; +import org.apache.pulsar.broker.service.Topic; +import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes; + +public class OpenTelemetryProducerStats implements AutoCloseable { + + // Replaces pulsar_producer_msg_rate_in + public static final String MESSAGE_IN_COUNTER = "pulsar.broker.producer.message.incoming.count"; + // Replaces pulsar_producer_msg_throughput_in + public static final String BYTES_IN_COUNTER = "pulsar.broker.producer.message.incoming.size"; + public static final String MESSAGE_DROP_COUNTER = "pulsar.broker.producer.message.drop.count"; + private final ObservableLongMeasurement messageInCounter; + private final ObservableLongMeasurement bytesInCounter; + private final ObservableLongMeasurement messageDropCounter; + + private final BatchCallback batchCallback; + + public OpenTelemetryProducerStats(PulsarService pulsar) { + var meter = pulsar.getOpenTelemetry().getMeter(); + + messageInCounter = meter + .counterBuilder(MESSAGE_IN_COUNTER) + .setUnit("{message}") + .setDescription("The total number of messages received from this producer.") + .buildObserver(); + + bytesInCounter = meter + .counterBuilder(BYTES_IN_COUNTER) + .setUnit("By") + .setDescription("The total number of messages bytes received from this producer.") + .buildObserver(); + + messageDropCounter = meter + .counterBuilder(MESSAGE_DROP_COUNTER) + .setUnit("{message}") + .setDescription("The total number of messages dropped from this producer.") + .buildObserver(); + + batchCallback = meter.batchCallback(() -> pulsar.getBrokerService() + .getTopics() + .values() + .stream() + .map(topicFuture -> topicFuture.getNow(Optional.empty())) + .filter(Optional::isPresent) + .map(Optional::get) + .map(Topic::getProducers) + .flatMap(p -> p.values().stream()).forEach(this::recordMetricsForProducer), + messageInCounter, + bytesInCounter, + messageDropCounter); + } + + @Override + public void close() { + batchCallback.close(); + } + + private void recordMetricsForProducer(Producer producer) { + var topicName = TopicName.get(producer.getTopic().getName()); + + var builder = Attributes.builder() + .put(OpenTelemetryAttributes.PULSAR_PRODUCER_NAME, producer.getProducerName()) + .put(OpenTelemetryAttributes.PULSAR_PRODUCER_ID, producer.getProducerId()) + .put(OpenTelemetryAttributes.PULSAR_PRODUCER_ACCESS_MODE, producer.getAccessMode().toString()) + .put(OpenTelemetryAttributes.PULSAR_PRODUCER_CONNECTED_SINCE, + producer.getConnectedSince().getEpochSecond()) + .put(OpenTelemetryAttributes.PULSAR_DOMAIN, topicName.getDomain().toString()) + .put(OpenTelemetryAttributes.PULSAR_TENANT, topicName.getTenant()) + .put(OpenTelemetryAttributes.PULSAR_NAMESPACE, topicName.getNamespace()) + .put(OpenTelemetryAttributes.PULSAR_TOPIC, topicName.getPartitionedTopicName()); + if (topicName.isPartitioned()) { + builder.put(OpenTelemetryAttributes.PULSAR_PARTITION_INDEX, topicName.getPartitionIndex()); + } + + var clientAddress = producer.getClientAddressAndPort(); + if (clientAddress != null) { + builder.put(OpenTelemetryAttributes.PULSAR_CLIENT_ADDRESS, clientAddress); + } + var clientVersion = producer.getClientVersion(); + if (clientVersion != null) { + builder.put(OpenTelemetryAttributes.PULSAR_CLIENT_VERSION, clientVersion); + } + var metadataList = producer.getMetadata() + .entrySet() + .stream() + .map(e -> String.format("%s:%s", e.getKey(), e.getValue())) + .toList(); + builder.put(OpenTelemetryAttributes.PULSAR_PRODUCER_METADATA, metadataList); + var attributes = builder.build(); + + var dummyValue = 0; + messageInCounter.record(dummyValue, attributes); + bytesInCounter.record(dummyValue, attributes); + messageDropCounter.record(dummyValue, attributes); + } +} From d76e9ea0d2f75f44be7b027cd206892c0235539a Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Mon, 20 May 2024 12:50:20 -0700 Subject: [PATCH 04/21] Update producer metric sources --- .../broker/stats/OpenTelemetryProducerStats.java | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java index 987898338d10f..1c8a83800b1ad 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java @@ -26,6 +26,7 @@ import org.apache.pulsar.broker.service.Producer; import org.apache.pulsar.broker.service.Topic; import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.common.policies.data.stats.NonPersistentPublisherStatsImpl; import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes; public class OpenTelemetryProducerStats implements AutoCloseable { @@ -33,7 +34,7 @@ public class OpenTelemetryProducerStats implements AutoCloseable { // Replaces pulsar_producer_msg_rate_in public static final String MESSAGE_IN_COUNTER = "pulsar.broker.producer.message.incoming.count"; // Replaces pulsar_producer_msg_throughput_in - public static final String BYTES_IN_COUNTER = "pulsar.broker.producer.message.incoming.size"; + public static final String BYTES_IN_COUNTER = "pulsar.broker.consumer.message.incoming.size"; public static final String MESSAGE_DROP_COUNTER = "pulsar.broker.producer.message.drop.count"; private final ObservableLongMeasurement messageInCounter; private final ObservableLongMeasurement bytesInCounter; @@ -114,9 +115,12 @@ private void recordMetricsForProducer(Producer producer) { builder.put(OpenTelemetryAttributes.PULSAR_PRODUCER_METADATA, metadataList); var attributes = builder.build(); - var dummyValue = 0; - messageInCounter.record(dummyValue, attributes); - bytesInCounter.record(dummyValue, attributes); - messageDropCounter.record(dummyValue, attributes); + var stats = producer.getStats(); + messageInCounter.record(stats.getMsgInCounter(), attributes); + bytesInCounter.record(stats.getBytesInCounter(), attributes); + + if (stats instanceof NonPersistentPublisherStatsImpl nonPersistentStats) { + messageDropCounter.record(nonPersistentStats.getMsgDropCount(), attributes); + } } } From 6b2883a4e63316e0f8c0a33333c763ad3baeca33 Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Mon, 20 May 2024 17:14:21 -0700 Subject: [PATCH 05/21] Add OpenTelemetryProducerStats field to PulsarService --- .../main/java/org/apache/pulsar/broker/PulsarService.java | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java index 6ee35ad295fb5..a1c678f3b5909 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java @@ -110,6 +110,7 @@ import org.apache.pulsar.broker.service.schema.SchemaStorageFactory; import org.apache.pulsar.broker.stats.MetricsGenerator; import org.apache.pulsar.broker.stats.OpenTelemetryConsumerStats; +import org.apache.pulsar.broker.stats.OpenTelemetryProducerStats; import org.apache.pulsar.broker.stats.OpenTelemetryTopicStats; import org.apache.pulsar.broker.stats.PulsarBrokerOpenTelemetry; import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsServlet; @@ -256,6 +257,7 @@ public class PulsarService implements AutoCloseable, ShutdownService { private final PulsarBrokerOpenTelemetry openTelemetry; private OpenTelemetryTopicStats openTelemetryTopicStats; private OpenTelemetryConsumerStats openTelemetryConsumerStats; + private OpenTelemetryProducerStats openTelemetryProducerStats; private TransactionMetadataStoreService transactionMetadataStoreService; private TransactionBufferProvider transactionBufferProvider; @@ -632,6 +634,10 @@ public CompletableFuture closeAsync() { brokerClientSharedTimer.stop(); monotonicSnapshotClock.close(); + if (openTelemetryProducerStats != null) { + openTelemetryProducerStats.close(); + openTelemetryProducerStats = null; + } if (openTelemetryConsumerStats != null) { openTelemetryConsumerStats.close(); openTelemetryConsumerStats = null; @@ -783,6 +789,7 @@ public void start() throws PulsarServerException { openTelemetryTopicStats = new OpenTelemetryTopicStats(this); openTelemetryConsumerStats = new OpenTelemetryConsumerStats(this); + openTelemetryProducerStats = new OpenTelemetryProducerStats(this); localMetadataSynchronizer = StringUtils.isNotBlank(config.getMetadataSyncEventTopic()) ? new PulsarMetadataEventSynchronizer(this, config.getMetadataSyncEventTopic()) From 6316c26bfe8b26250de9e0d2e6bb49dbb37a4d99 Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Fri, 7 Jun 2024 10:21:37 -0700 Subject: [PATCH 06/21] Cosmetic fixes --- .../pulsar/broker/stats/OpenTelemetryProducerStats.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java index 1c8a83800b1ad..f68ab2806c921 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java @@ -21,6 +21,8 @@ import io.opentelemetry.api.common.Attributes; import io.opentelemetry.api.metrics.BatchCallback; import io.opentelemetry.api.metrics.ObservableLongMeasurement; +import java.util.Collection; +import java.util.Map; import java.util.Optional; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.service.Producer; @@ -71,7 +73,9 @@ public OpenTelemetryProducerStats(PulsarService pulsar) { .filter(Optional::isPresent) .map(Optional::get) .map(Topic::getProducers) - .flatMap(p -> p.values().stream()).forEach(this::recordMetricsForProducer), + .map(Map::values) + .flatMap(Collection::stream) + .forEach(this::recordMetricsForProducer), messageInCounter, bytesInCounter, messageDropCounter); From 70bf82f94a80c1de926ec1a78b9398abe91a63cb Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Fri, 7 Jun 2024 10:23:17 -0700 Subject: [PATCH 07/21] Reduce producer attribute count --- .../stats/OpenTelemetryProducerStats.java | 17 ----------------- .../opentelemetry/OpenTelemetryAttributes.java | 11 +---------- 2 files changed, 1 insertion(+), 27 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java index f68ab2806c921..fc6484b332387 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java @@ -93,8 +93,6 @@ private void recordMetricsForProducer(Producer producer) { .put(OpenTelemetryAttributes.PULSAR_PRODUCER_NAME, producer.getProducerName()) .put(OpenTelemetryAttributes.PULSAR_PRODUCER_ID, producer.getProducerId()) .put(OpenTelemetryAttributes.PULSAR_PRODUCER_ACCESS_MODE, producer.getAccessMode().toString()) - .put(OpenTelemetryAttributes.PULSAR_PRODUCER_CONNECTED_SINCE, - producer.getConnectedSince().getEpochSecond()) .put(OpenTelemetryAttributes.PULSAR_DOMAIN, topicName.getDomain().toString()) .put(OpenTelemetryAttributes.PULSAR_TENANT, topicName.getTenant()) .put(OpenTelemetryAttributes.PULSAR_NAMESPACE, topicName.getNamespace()) @@ -102,21 +100,6 @@ private void recordMetricsForProducer(Producer producer) { if (topicName.isPartitioned()) { builder.put(OpenTelemetryAttributes.PULSAR_PARTITION_INDEX, topicName.getPartitionIndex()); } - - var clientAddress = producer.getClientAddressAndPort(); - if (clientAddress != null) { - builder.put(OpenTelemetryAttributes.PULSAR_CLIENT_ADDRESS, clientAddress); - } - var clientVersion = producer.getClientVersion(); - if (clientVersion != null) { - builder.put(OpenTelemetryAttributes.PULSAR_CLIENT_VERSION, clientVersion); - } - var metadataList = producer.getMetadata() - .entrySet() - .stream() - .map(e -> String.format("%s:%s", e.getKey(), e.getValue())) - .toList(); - builder.put(OpenTelemetryAttributes.PULSAR_PRODUCER_METADATA, metadataList); var attributes = builder.build(); var stats = producer.getStats(); diff --git a/pulsar-opentelemetry/src/main/java/org/apache/pulsar/opentelemetry/OpenTelemetryAttributes.java b/pulsar-opentelemetry/src/main/java/org/apache/pulsar/opentelemetry/OpenTelemetryAttributes.java index e9860c1a167ed..47c2c78e18fa1 100644 --- a/pulsar-opentelemetry/src/main/java/org/apache/pulsar/opentelemetry/OpenTelemetryAttributes.java +++ b/pulsar-opentelemetry/src/main/java/org/apache/pulsar/opentelemetry/OpenTelemetryAttributes.java @@ -19,6 +19,7 @@ package org.apache.pulsar.opentelemetry; import io.opentelemetry.api.common.AttributeKey; + import java.util.List; /** @@ -101,16 +102,6 @@ public interface OpenTelemetryAttributes { */ AttributeKey PULSAR_PRODUCER_ACCESS_MODE = AttributeKey.stringKey("pulsar.producer.access_mode"); - /** - * The producer metadata properties, as a list of "key:value" pairs. - */ - AttributeKey> PULSAR_PRODUCER_METADATA = AttributeKey.stringArrayKey("pulsar.producer.metadata"); - - /** - * The UTC timestamp of the Pulsar producer creation. - */ - AttributeKey PULSAR_PRODUCER_CONNECTED_SINCE = AttributeKey.longKey("pulsar.producer.connected_since"); - /** * The address of the Pulsar client. */ From 4226aa4048f97c800c162898cbbf60ff333be7da Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Fri, 7 Jun 2024 10:31:12 -0700 Subject: [PATCH 08/21] Cache attributes --- .../pulsar/broker/service/Producer.java | 31 +++++++++++++++++++ .../stats/OpenTelemetryProducerStats.java | 20 ++---------- .../OpenTelemetryAttributes.java | 1 - 3 files changed, 33 insertions(+), 19 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java index 90b852ee73d7b..566cf65ec2816 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java @@ -27,6 +27,7 @@ import io.netty.buffer.ByteBuf; import io.netty.util.Recycler; import io.netty.util.Recycler.Handle; +import io.opentelemetry.api.common.Attributes; import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -36,6 +37,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLongFieldUpdater; +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; import org.apache.bookkeeper.mledger.Position; import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.intercept.BrokerInterceptor; @@ -59,6 +61,7 @@ import org.apache.pulsar.common.protocol.schema.SchemaVersion; import org.apache.pulsar.common.stats.Rate; import org.apache.pulsar.common.util.DateFormatter; +import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -90,6 +93,10 @@ public class Producer { private final CompletableFuture closeFuture; private final PublisherStatsImpl stats; + private volatile Attributes attributes = null; + private static final AtomicReferenceFieldUpdater ATTRIBUTES_FIELD_UPDATER = + AtomicReferenceFieldUpdater.newUpdater(Producer.class, Attributes.class, "attributes"); + private final boolean isRemote; private final String remoteCluster; private final boolean isNonPersistentTopic; @@ -882,4 +889,28 @@ public void incrementThrottleCount() { public void decrementThrottleCount() { cnx.decrementThrottleCount(); } + + public Attributes getOpenTelemetryAttributes() { + if (attributes != null) { + return attributes; + } + return ATTRIBUTES_FIELD_UPDATER.updateAndGet(this, old -> { + if (old != null) { + return old; + } + var topicName = TopicName.get(topic.getName()); + var builder = Attributes.builder() + .put(OpenTelemetryAttributes.PULSAR_PRODUCER_NAME, producerName) + .put(OpenTelemetryAttributes.PULSAR_PRODUCER_ID, producerId) + .put(OpenTelemetryAttributes.PULSAR_PRODUCER_ACCESS_MODE, accessMode.name().toLowerCase()) + .put(OpenTelemetryAttributes.PULSAR_DOMAIN, topicName.getDomain().toString()) + .put(OpenTelemetryAttributes.PULSAR_TENANT, topicName.getTenant()) + .put(OpenTelemetryAttributes.PULSAR_NAMESPACE, topicName.getNamespace()) + .put(OpenTelemetryAttributes.PULSAR_TOPIC, topicName.getPartitionedTopicName()); + if (topicName.isPartitioned()) { + builder.put(OpenTelemetryAttributes.PULSAR_PARTITION_INDEX, topicName.getPartitionIndex()); + } + return builder.build(); + }); + } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java index fc6484b332387..0e50f7467e284 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java @@ -18,7 +18,6 @@ */ package org.apache.pulsar.broker.stats; -import io.opentelemetry.api.common.Attributes; import io.opentelemetry.api.metrics.BatchCallback; import io.opentelemetry.api.metrics.ObservableLongMeasurement; import java.util.Collection; @@ -27,9 +26,7 @@ import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.service.Producer; import org.apache.pulsar.broker.service.Topic; -import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.policies.data.stats.NonPersistentPublisherStatsImpl; -import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes; public class OpenTelemetryProducerStats implements AutoCloseable { @@ -87,22 +84,9 @@ public void close() { } private void recordMetricsForProducer(Producer producer) { - var topicName = TopicName.get(producer.getTopic().getName()); - - var builder = Attributes.builder() - .put(OpenTelemetryAttributes.PULSAR_PRODUCER_NAME, producer.getProducerName()) - .put(OpenTelemetryAttributes.PULSAR_PRODUCER_ID, producer.getProducerId()) - .put(OpenTelemetryAttributes.PULSAR_PRODUCER_ACCESS_MODE, producer.getAccessMode().toString()) - .put(OpenTelemetryAttributes.PULSAR_DOMAIN, topicName.getDomain().toString()) - .put(OpenTelemetryAttributes.PULSAR_TENANT, topicName.getTenant()) - .put(OpenTelemetryAttributes.PULSAR_NAMESPACE, topicName.getNamespace()) - .put(OpenTelemetryAttributes.PULSAR_TOPIC, topicName.getPartitionedTopicName()); - if (topicName.isPartitioned()) { - builder.put(OpenTelemetryAttributes.PULSAR_PARTITION_INDEX, topicName.getPartitionIndex()); - } - var attributes = builder.build(); - + var attributes = producer.getOpenTelemetryAttributes(); var stats = producer.getStats(); + messageInCounter.record(stats.getMsgInCounter(), attributes); bytesInCounter.record(stats.getBytesInCounter(), attributes); diff --git a/pulsar-opentelemetry/src/main/java/org/apache/pulsar/opentelemetry/OpenTelemetryAttributes.java b/pulsar-opentelemetry/src/main/java/org/apache/pulsar/opentelemetry/OpenTelemetryAttributes.java index 47c2c78e18fa1..64919099e66e4 100644 --- a/pulsar-opentelemetry/src/main/java/org/apache/pulsar/opentelemetry/OpenTelemetryAttributes.java +++ b/pulsar-opentelemetry/src/main/java/org/apache/pulsar/opentelemetry/OpenTelemetryAttributes.java @@ -19,7 +19,6 @@ package org.apache.pulsar.opentelemetry; import io.opentelemetry.api.common.AttributeKey; - import java.util.List; /** From 4826cfb06d3aeefeba20bf74f13a1b58e34038ae Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Fri, 7 Jun 2024 11:25:14 -0700 Subject: [PATCH 09/21] Add producer stats test --- .../stats/OpenTelemetryProducerStatsTest.java | 88 +++++++++++++++++++ 1 file changed, 88 insertions(+) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStatsTest.java diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStatsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStatsTest.java new file mode 100644 index 0000000000000..e273ac4446141 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStatsTest.java @@ -0,0 +1,88 @@ +/* + * 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 static org.apache.pulsar.broker.stats.BrokerOpenTelemetryTestUtil.assertMetricLongSumValue; +import static org.assertj.core.api.Assertions.assertThat; +import io.opentelemetry.api.common.Attributes; +import lombok.Cleanup; +import org.apache.pulsar.broker.BrokerTestUtil; +import org.apache.pulsar.broker.service.BrokerTestBase; +import org.apache.pulsar.broker.testcontext.PulsarTestContext; +import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +public class OpenTelemetryProducerStatsTest extends BrokerTestBase { + + @BeforeMethod(alwaysRun = true) + @Override + protected void setup() throws Exception { + super.baseSetup(); + } + + @AfterMethod(alwaysRun = true) + @Override + protected void cleanup() throws Exception { + super.internalCleanup(); + } + + @Override + protected void customizeMainPulsarTestContextBuilder(PulsarTestContext.Builder builder) { + super.customizeMainPulsarTestContextBuilder(builder); + builder.enableOpenTelemetry(true); + } + + + @Test(timeOut = 30_000) + public void testMessagingMetrics() throws Exception { + var topicName = BrokerTestUtil.newUniqueName("persistent://prop/ns-abc/testProducerMessagingMetrics"); + admin.topics().createNonPartitionedTopic(topicName); + + var messageCount = 5; + var producerName = BrokerTestUtil.newUniqueName("testProducerName"); + + @Cleanup + var producer = pulsarClient.newProducer() + .producerName(producerName) + .topic(topicName) + .create(); + for (int i = 0; i < messageCount; i++) { + producer.send(String.format("msg-%d", i).getBytes()); + } + + var attributes = Attributes.builder() + .put(OpenTelemetryAttributes.PULSAR_DOMAIN, "persistent") + .put(OpenTelemetryAttributes.PULSAR_TENANT, "prop") + .put(OpenTelemetryAttributes.PULSAR_NAMESPACE, "prop/ns-abc") + .put(OpenTelemetryAttributes.PULSAR_TOPIC, topicName) + .put(OpenTelemetryAttributes.PULSAR_PRODUCER_NAME, producerName) + .put(OpenTelemetryAttributes.PULSAR_PRODUCER_ID, 0) + .put(OpenTelemetryAttributes.PULSAR_PRODUCER_ACCESS_MODE, "shared") + .build(); + + var metrics = pulsarTestContext.getOpenTelemetryMetricReader().collectAllMetrics(); + + assertMetricLongSumValue(metrics, OpenTelemetryProducerStats.MESSAGE_IN_COUNTER, attributes, + actual -> assertThat(actual).isPositive()); + assertMetricLongSumValue(metrics, OpenTelemetryProducerStats.BYTES_IN_COUNTER, attributes, + actual -> assertThat(actual).isPositive()); + } +} From 92b2215441f6eceebd56e8ffcf720da81da7d3d0 Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Fri, 7 Jun 2024 11:39:58 -0700 Subject: [PATCH 10/21] Update metric defs --- .../stats/OpenTelemetryProducerStats.java | 21 +++++++++---------- 1 file changed, 10 insertions(+), 11 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java index 0e50f7467e284..78a8067743644 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java @@ -20,23 +20,24 @@ import io.opentelemetry.api.metrics.BatchCallback; import io.opentelemetry.api.metrics.ObservableLongMeasurement; -import java.util.Collection; -import java.util.Map; import java.util.Optional; +import java.util.concurrent.CompletableFuture; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.service.Producer; -import org.apache.pulsar.broker.service.Topic; import org.apache.pulsar.common.policies.data.stats.NonPersistentPublisherStatsImpl; public class OpenTelemetryProducerStats implements AutoCloseable { // Replaces pulsar_producer_msg_rate_in public static final String MESSAGE_IN_COUNTER = "pulsar.broker.producer.message.incoming.count"; - // Replaces pulsar_producer_msg_throughput_in - public static final String BYTES_IN_COUNTER = "pulsar.broker.consumer.message.incoming.size"; - public static final String MESSAGE_DROP_COUNTER = "pulsar.broker.producer.message.drop.count"; private final ObservableLongMeasurement messageInCounter; + + // Replaces pulsar_producer_msg_throughput_in + public static final String BYTES_IN_COUNTER = "pulsar.broker.producer.message.incoming.size"; private final ObservableLongMeasurement bytesInCounter; + + // Replaces pulsar_consumer_msg_ack_rate + public static final String MESSAGE_DROP_COUNTER = "pulsar.broker.producer.message.drop.count"; private final ObservableLongMeasurement messageDropCounter; private final BatchCallback batchCallback; @@ -66,12 +67,10 @@ public OpenTelemetryProducerStats(PulsarService pulsar) { .getTopics() .values() .stream() - .map(topicFuture -> topicFuture.getNow(Optional.empty())) + .filter(future -> future.isDone() && !future.isCompletedExceptionally()) + .map(CompletableFuture::join) .filter(Optional::isPresent) - .map(Optional::get) - .map(Topic::getProducers) - .map(Map::values) - .flatMap(Collection::stream) + .flatMap(topic -> topic.get().getProducers().values().stream()) .forEach(this::recordMetricsForProducer), messageInCounter, bytesInCounter, From d784728735bfb08e3df456cd29ed96f2dc7303a4 Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Fri, 7 Jun 2024 12:08:02 -0700 Subject: [PATCH 11/21] Add totalValue field to Rate --- .../src/main/java/org/apache/pulsar/common/stats/Rate.java | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/stats/Rate.java b/pulsar-common/src/main/java/org/apache/pulsar/common/stats/Rate.java index 886e31ab71216..936962d8ee544 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/stats/Rate.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/stats/Rate.java @@ -28,6 +28,7 @@ public class Rate { private final LongAdder valueAdder = new LongAdder(); private final LongAdder countAdder = new LongAdder(); private final LongAdder totalCountAdder = new LongAdder(); + private final LongAdder totalValueAdder = new LongAdder(); // Computed stats private long count = 0L; @@ -43,12 +44,14 @@ public void recordEvent() { public void recordEvent(long value) { valueAdder.add(value); + totalValueAdder.add(value); countAdder.increment(); totalCountAdder.increment(); } public void recordMultipleEvents(long events, long totalValue) { valueAdder.add(totalValue); + totalValueAdder.add(totalValue); countAdder.add(events); totalCountAdder.add(events); } @@ -88,4 +91,8 @@ public double getValueRate() { public long getTotalCount() { return this.totalCountAdder.longValue(); } + + public long getTotalValue() { + return this.totalValueAdder.sum(); + } } From 818428392a7d377f3f9865a411e75fe409e03858 Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Fri, 7 Jun 2024 12:25:54 -0700 Subject: [PATCH 12/21] Revert interface changes --- .../common/policies/data/NonPersistentPublisherStats.java | 3 --- .../apache/pulsar/common/policies/data/PublisherStats.java | 7 ------- 2 files changed, 10 deletions(-) diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/NonPersistentPublisherStats.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/NonPersistentPublisherStats.java index ce82724cf4c0b..fec645c2e258d 100644 --- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/NonPersistentPublisherStats.java +++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/NonPersistentPublisherStats.java @@ -27,7 +27,4 @@ public interface NonPersistentPublisherStats extends PublisherStats { * messages per connection. **/ double getMsgDropRate(); - - void recordMsgDrop(long numMessages); - long getMsgDropCount(); } \ No newline at end of file diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/PublisherStats.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/PublisherStats.java index 888d4642ea315..78af224d28c02 100644 --- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/PublisherStats.java +++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/PublisherStats.java @@ -60,11 +60,4 @@ public interface PublisherStats { /** Metadata (key/value strings) associated with this publisher. */ Map getMetadata(); - - void recordMsgIn(long messageCount, long byteCount); - long getMsgInCounter(); - long getBytesInCounter(); - - void recordChunkedMsgIn(long messageCount); - long getChunkedMsgIn(); } From 8966aefe9571f275eb744c39dd7dbf6fc1eee5fc Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Fri, 7 Jun 2024 12:28:24 -0700 Subject: [PATCH 13/21] Relocate Producer.rateIn field to PublisherStatsImpl --- .../pulsar/broker/service/Producer.java | 32 +++++-------------- .../data/stats/PublisherStatsImpl.java | 27 +++++++++------- 2 files changed, 23 insertions(+), 36 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java index 566cf65ec2816..1b2893ecd9f6f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java @@ -78,8 +78,6 @@ public class Producer { private final String appId; private final BrokerInterceptor brokerInterceptor; @Deprecated - private Rate msgIn; - @Deprecated private Rate chunkedMessageRate; // it records msg-drop rate only for non-persistent topic @Deprecated @@ -128,7 +126,6 @@ public Producer(Topic topic, TransportCnx cnx, long producerId, String producerN this.epoch = epoch; this.closeFuture = new CompletableFuture<>(); this.appId = appId; - this.msgIn = new Rate(); this.chunkedMessageRate = new Rate(); this.isNonPersistentTopic = topic instanceof NonPersistentTopic; this.msgDrop = this.isNonPersistentTopic ? new Rate() : null; @@ -280,7 +277,7 @@ public boolean checkAndStartPublish(long producerId, long sequenceId, ByteBuf he private void publishMessageToTopic(ByteBuf headersAndPayload, long sequenceId, long batchSize, boolean isChunked, boolean isMarker, Position position) { MessagePublishContext messagePublishContext = - MessagePublishContext.get(this, sequenceId, msgIn, headersAndPayload.readableBytes(), + MessagePublishContext.get(this, sequenceId, headersAndPayload.readableBytes(), batchSize, isChunked, System.nanoTime(), isMarker, position); if (brokerInterceptor != null) { brokerInterceptor @@ -292,7 +289,7 @@ private void publishMessageToTopic(ByteBuf headersAndPayload, long sequenceId, l private void publishMessageToTopic(ByteBuf headersAndPayload, long lowestSequenceId, long highestSequenceId, long batchSize, boolean isChunked, boolean isMarker, Position position) { MessagePublishContext messagePublishContext = MessagePublishContext.get(this, lowestSequenceId, - highestSequenceId, msgIn, headersAndPayload.readableBytes(), batchSize, + highestSequenceId, headersAndPayload.readableBytes(), batchSize, isChunked, System.nanoTime(), isMarker, position); if (brokerInterceptor != null) { brokerInterceptor @@ -384,8 +381,6 @@ private static final class MessagePublishContext implements PublishContext, Runn private long sequenceId; private long ledgerId; private long entryId; - @Deprecated - private Rate rateIn; private int msgSize; private long batchSize; private boolean chunked; @@ -552,7 +547,6 @@ public void run() { } // stats - rateIn.recordMultipleEvents(batchSize, msgSize); producer.stats.recordMsgIn(batchSize, msgSize); producer.topic.recordAddLatency(System.nanoTime() - startTimeNs, TimeUnit.NANOSECONDS); producer.cnx.getCommandSender().sendSendReceiptResponse(producer.producerId, sequenceId, highestSequenceId, @@ -570,12 +564,11 @@ public void run() { recycle(); } - static MessagePublishContext get(Producer producer, long sequenceId, Rate rateIn, int msgSize, - long batchSize, boolean chunked, long startTimeNs, boolean isMarker, Position position) { + static MessagePublishContext get(Producer producer, long sequenceId, int msgSize, long batchSize, + boolean chunked, long startTimeNs, boolean isMarker, Position position) { MessagePublishContext callback = RECYCLER.get(); callback.producer = producer; callback.sequenceId = sequenceId; - callback.rateIn = rateIn; callback.msgSize = msgSize; callback.batchSize = batchSize; callback.chunked = chunked; @@ -591,13 +584,12 @@ static MessagePublishContext get(Producer producer, long sequenceId, Rate rateIn return callback; } - static MessagePublishContext get(Producer producer, long lowestSequenceId, long highestSequenceId, Rate rateIn, - int msgSize, long batchSize, boolean chunked, long startTimeNs, boolean isMarker, Position position) { + static MessagePublishContext get(Producer producer, long lowestSequenceId, long highestSequenceId, int msgSize, + long batchSize, boolean chunked, long startTimeNs, boolean isMarker, Position position) { MessagePublishContext callback = RECYCLER.get(); callback.producer = producer; callback.sequenceId = lowestSequenceId; callback.highestSequenceId = highestSequenceId; - callback.rateIn = rateIn; callback.msgSize = msgSize; callback.batchSize = batchSize; callback.originalProducerName = null; @@ -646,7 +638,6 @@ public void recycle() { highestSequenceId = -1L; originalSequenceId = -1L; originalHighestSequenceId = -1L; - rateIn = null; msgSize = 0; ledgerId = -1L; entryId = -1L; @@ -751,14 +742,7 @@ public void topicMigrated(Optional clusterUrl) { } public void updateRates() { - msgIn.calculateRate(); - chunkedMessageRate.calculateRate(); - stats.msgRateIn = msgIn.getRate(); - stats.msgThroughputIn = msgIn.getValueRate(); - stats.averageMsgSize = msgIn.getAverageValue(); - stats.chunkedMessageRate = chunkedMessageRate.getRate(); - if (chunkedMessageRate.getCount() > 0 && this.topic instanceof PersistentTopic) { - ((PersistentTopic) this.topic).msgChunkPublished = true; + stats.calculateRates(); } if (this.isNonPersistentTopic) { msgDrop.calculateRate(); @@ -835,7 +819,7 @@ public void publishTxnMessage(TxnID txnID, long producerId, long sequenceId, lon return; } MessagePublishContext messagePublishContext = - MessagePublishContext.get(this, sequenceId, highSequenceId, msgIn, + MessagePublishContext.get(this, sequenceId, highSequenceId, headersAndPayload.readableBytes(), batchSize, isChunked, System.nanoTime(), isMarker, null); if (brokerInterceptor != null) { brokerInterceptor diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java index db3a3a8469fc9..2dbadeea8753e 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java @@ -20,10 +20,10 @@ import com.fasterxml.jackson.annotation.JsonIgnore; import java.util.Map; -import java.util.concurrent.atomic.LongAdder; import lombok.Data; import org.apache.pulsar.client.api.ProducerAccessMode; import org.apache.pulsar.common.policies.data.PublisherStats; +import org.apache.pulsar.common.stats.Rate; /** * Statistics about a publisher. @@ -35,10 +35,6 @@ public class PublisherStatsImpl implements PublisherStats { public ProducerAccessMode accessMode; - private final LongAdder msgInCounter = new LongAdder(); - private final LongAdder bytesInCounter = new LongAdder(); - private final LongAdder chunkedMessageCounter = new LongAdder(); - /** Total rate of messages published by this publisher (msg/s). */ public double msgRateIn; @@ -69,6 +65,8 @@ public class PublisherStatsImpl implements PublisherStats { /** Metadata (key/value strings) associated with this publisher. */ public Map metadata; + @JsonIgnore + private final Rate msgIn = new Rate(); public PublisherStatsImpl add(PublisherStatsImpl stats) { if (stats == null) { throw new IllegalArgumentException("stats can't be null"); @@ -113,20 +111,25 @@ public void setClientVersion(String clientVersion) { this.clientVersion = clientVersion; } - @Override + public void calculateRates() { + msgIn.calculateRate(); + + msgRateIn = msgIn.getRate(); + msgThroughputIn = msgIn.getValueRate(); + averageMsgSize = msgIn.getAverageValue(); + chunkedMessageRate = msgChunkIn.getRate(); + } + public void recordMsgIn(long messageCount, long byteCount) { - msgInCounter.add(messageCount); - bytesInCounter.add(byteCount); + msgIn.recordMultipleEvents(messageCount, byteCount); } - @Override public long getMsgInCounter() { - return msgInCounter.sum(); + return msgIn.getTotalCount(); } - @Override public long getBytesInCounter() { - return bytesInCounter.sum(); + return msgIn.getTotalValue(); } @Override From eb24b9ab3ab5f44f8e88e0e27ed30afd0d496aea Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Fri, 7 Jun 2024 12:28:40 -0700 Subject: [PATCH 14/21] Remove unused method Producer.updateRates --- .../main/java/org/apache/pulsar/broker/service/Producer.java | 3 --- 1 file changed, 3 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java index 1b2893ecd9f6f..dc42b74c1f6cd 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java @@ -749,9 +749,6 @@ public void updateRates() { ((NonPersistentPublisherStatsImpl) stats).msgDropRate = msgDrop.getValueRate(); } } - - public void updateRates(int numOfMessages, long msgSizeInBytes) { - msgIn.recordMultipleEvents(numOfMessages, msgSizeInBytes); } public boolean isRemote() { From cb3f2365ff0354f20e02b7a3aa04c907af7643be Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Fri, 7 Jun 2024 12:30:13 -0700 Subject: [PATCH 15/21] Relocate Producer.chunkedMessageRate to PublisherStatsImpl --- .../org/apache/pulsar/broker/service/Producer.java | 8 +++----- .../policies/data/stats/PublisherStatsImpl.java | 14 ++++++++------ 2 files changed, 11 insertions(+), 11 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java index dc42b74c1f6cd..e33f34157317d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java @@ -77,8 +77,6 @@ public class Producer { private final long producerId; private final String appId; private final BrokerInterceptor brokerInterceptor; - @Deprecated - private Rate chunkedMessageRate; // it records msg-drop rate only for non-persistent topic @Deprecated private final Rate msgDrop; @@ -126,7 +124,6 @@ public Producer(Topic topic, TransportCnx cnx, long producerId, String producerN this.epoch = epoch; this.closeFuture = new CompletableFuture<>(); this.appId = appId; - this.chunkedMessageRate = new Rate(); this.isNonPersistentTopic = topic instanceof NonPersistentTopic; this.msgDrop = this.isNonPersistentTopic ? new Rate() : null; this.isShadowTopic = @@ -553,8 +550,7 @@ public void run() { ledgerId, entryId); producer.cnx.completedSendOperation(producer.isNonPersistentTopic, msgSize); if (this.chunked) { - producer.chunkedMessageRate.recordEvent(); - producer.stats.recordChunkedMsgIn(1); + producer.stats.recordChunkedMsgIn(); } producer.publishOperationCompleted(); if (producer.brokerInterceptor != null) { @@ -743,6 +739,8 @@ public void topicMigrated(Optional clusterUrl) { public void updateRates() { stats.calculateRates(); + if (stats.getMsgChunkIn().getCount() > 0 && topic instanceof PersistentTopic persistentTopic) { + persistentTopic.msgChunkPublished = true; } if (this.isNonPersistentTopic) { msgDrop.calculateRate(); diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java index 2dbadeea8753e..9326311d67cc2 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java @@ -67,6 +67,9 @@ public class PublisherStatsImpl implements PublisherStats { @JsonIgnore private final Rate msgIn = new Rate(); + @JsonIgnore + private final Rate msgChunkIn = new Rate(); + public PublisherStatsImpl add(PublisherStatsImpl stats) { if (stats == null) { throw new IllegalArgumentException("stats can't be null"); @@ -113,6 +116,7 @@ public void setClientVersion(String clientVersion) { public void calculateRates() { msgIn.calculateRate(); + msgChunkIn.calculateRate(); msgRateIn = msgIn.getRate(); msgThroughputIn = msgIn.getValueRate(); @@ -132,13 +136,11 @@ public long getBytesInCounter() { return msgIn.getTotalValue(); } - @Override - public void recordChunkedMsgIn(long messageCount) { - chunkedMessageCounter.add(messageCount); + public void recordChunkedMsgIn() { + msgChunkIn.recordEvent(); } - @Override - public long getChunkedMsgIn() { - return chunkedMessageCounter.sum(); + public long getChunkedMsgInCounter() { + return msgChunkIn.getTotalCount(); } } From 5472bfe715c2d04aac61db92c9852eb4e8dc112a Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Fri, 7 Jun 2024 12:32:14 -0700 Subject: [PATCH 16/21] Relocate Producer.msgDropRate to NonPersistentPublisherStatsImpl --- .../apache/pulsar/broker/service/Producer.java | 14 ++------------ .../stats/NonPersistentPublisherStatsImpl.java | 18 ++++++++++++------ 2 files changed, 14 insertions(+), 18 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java index e33f34157317d..3d31f4e00fb45 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java @@ -59,7 +59,6 @@ import org.apache.pulsar.common.policies.data.stats.PublisherStatsImpl; import org.apache.pulsar.common.protocol.Commands; import org.apache.pulsar.common.protocol.schema.SchemaVersion; -import org.apache.pulsar.common.stats.Rate; import org.apache.pulsar.common.util.DateFormatter; import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes; import org.slf4j.Logger; @@ -77,9 +76,6 @@ public class Producer { private final long producerId; private final String appId; private final BrokerInterceptor brokerInterceptor; - // it records msg-drop rate only for non-persistent topic - @Deprecated - private final Rate msgDrop; private volatile long pendingPublishAcks = 0; private static final AtomicLongFieldUpdater pendingPublishAcksUpdater = AtomicLongFieldUpdater @@ -125,7 +121,6 @@ public Producer(Topic topic, TransportCnx cnx, long producerId, String producerN this.closeFuture = new CompletableFuture<>(); this.appId = appId; this.isNonPersistentTopic = topic instanceof NonPersistentTopic; - this.msgDrop = this.isNonPersistentTopic ? new Rate() : null; this.isShadowTopic = topic instanceof PersistentTopic && ((PersistentTopic) topic).getShadowSourceTopic().isPresent(); @@ -343,8 +338,8 @@ private void publishOperationCompleted() { } public void recordMessageDrop(int batchSize) { - if (this.isNonPersistentTopic) { - msgDrop.recordEvent(batchSize); + if (stats instanceof NonPersistentPublisherStatsImpl nonPersistentPublisherStats) { + nonPersistentPublisherStats.recordMsgDrop(batchSize); } } @@ -742,11 +737,6 @@ public void updateRates() { if (stats.getMsgChunkIn().getCount() > 0 && topic instanceof PersistentTopic persistentTopic) { persistentTopic.msgChunkPublished = true; } - if (this.isNonPersistentTopic) { - msgDrop.calculateRate(); - ((NonPersistentPublisherStatsImpl) stats).msgDropRate = msgDrop.getValueRate(); - } - } } public boolean isRemote() { diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/NonPersistentPublisherStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/NonPersistentPublisherStatsImpl.java index 66e94ace46d42..8f7fc01e64b7f 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/NonPersistentPublisherStatsImpl.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/NonPersistentPublisherStatsImpl.java @@ -18,11 +18,12 @@ */ package org.apache.pulsar.common.policies.data.stats; +import com.fasterxml.jackson.annotation.JsonIgnore; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.util.Objects; -import java.util.concurrent.atomic.LongAdder; import lombok.Getter; import org.apache.pulsar.common.policies.data.NonPersistentPublisherStats; +import org.apache.pulsar.common.stats.Rate; /** * Non-persistent publisher statistics. @@ -36,7 +37,8 @@ public class NonPersistentPublisherStatsImpl extends PublisherStatsImpl implemen @Getter public double msgDropRate; - private final LongAdder msgDropCount = new LongAdder(); + @JsonIgnore + private final Rate msgDrop = new Rate(); public NonPersistentPublisherStatsImpl add(NonPersistentPublisherStatsImpl stats) { Objects.requireNonNull(stats); @@ -45,13 +47,17 @@ public NonPersistentPublisherStatsImpl add(NonPersistentPublisherStatsImpl stats return this; } - @Override + public void calculateRates() { + super.calculateRates(); + msgDrop.calculateRate(); + msgDropRate = msgDrop.getRate(); + } + public void recordMsgDrop(long numMessages) { - msgDropCount.add(numMessages); + msgDrop.recordEvent(numMessages); } - @Override public long getMsgDropCount() { - return msgDropCount.sum(); + return msgDrop.getTotalCount(); } } From 65b0d1cbe1a9a8efc2655cd1fb469a99694b21d3 Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Fri, 7 Jun 2024 12:49:41 -0700 Subject: [PATCH 17/21] Validate message drop metric --- .../client/api/NonPersistentTopicTest.java | 31 +++++++++++++++++-- 1 file changed, 29 insertions(+), 2 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java index 4f64c4271fe89..e5c992ec6f858 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java @@ -18,6 +18,8 @@ */ package org.apache.pulsar.client.api; +import static org.apache.pulsar.broker.stats.BrokerOpenTelemetryTestUtil.assertMetricLongSumValue; +import static org.assertj.core.api.Assertions.assertThat; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertNotEquals; @@ -27,6 +29,7 @@ import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; import com.google.common.collect.Sets; +import io.opentelemetry.api.common.Attributes; import java.net.URL; import java.util.HashSet; import java.util.Optional; @@ -50,6 +53,8 @@ import org.apache.pulsar.broker.service.BrokerService; import org.apache.pulsar.broker.service.nonpersistent.NonPersistentReplicator; import org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic; +import org.apache.pulsar.broker.stats.OpenTelemetryProducerStats; +import org.apache.pulsar.broker.testcontext.PulsarTestContext; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.impl.ConsumerImpl; import org.apache.pulsar.client.impl.MessageIdImpl; @@ -65,6 +70,7 @@ import org.apache.pulsar.common.policies.data.SubscriptionStats; import org.apache.pulsar.common.policies.data.TenantInfoImpl; import org.apache.pulsar.common.policies.data.TopicType; +import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes; import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble; import org.apache.pulsar.zookeeper.ZookeeperServerTest; import org.awaitility.Awaitility; @@ -105,6 +111,12 @@ protected void cleanup() throws Exception { super.internalCleanup(); } + @Override + protected void customizeMainPulsarTestContextBuilder(PulsarTestContext.Builder pulsarTestContextBuilder) { + super.customizeMainPulsarTestContextBuilder(pulsarTestContextBuilder); + pulsarTestContextBuilder.enableOpenTelemetry(true); + } + @Test(timeOut = 90000 /* 1.5mn */) public void testNonPersistentPartitionsAreNotAutoCreatedWhenThePartitionedTopicDoesNotExist() throws Exception { final boolean defaultAllowAutoTopicCreation = conf.isAllowAutoTopicCreation(); @@ -357,9 +369,12 @@ public void testProducerRateLimit() throws Exception { @Cleanup("shutdownNow") ExecutorService executor = Executors.newFixedThreadPool(5); AtomicBoolean failed = new AtomicBoolean(false); + @Cleanup Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("subscriber-1") .subscribe(); - Producer producer = pulsarClient.newProducer().topic(topic).create(); + var producerName = BrokerTestUtil.newUniqueName("testProducerRateLimit"); + @Cleanup + Producer producer = pulsarClient.newProducer().topic(topic).producerName(producerName).create(); byte[] msgData = "testData".getBytes(); final int totalProduceMessages = 10; CountDownLatch latch = new CountDownLatch(totalProduceMessages); @@ -392,7 +407,19 @@ public void testProducerRateLimit() throws Exception { // but as message should be dropped at broker: broker should not receive the message assertNotEquals(messageSet.size(), totalProduceMessages); - producer.close(); + // Verify the corresponding metric is updated + var attributes = Attributes.builder() + .put(OpenTelemetryAttributes.PULSAR_PRODUCER_NAME, producerName) + .put(OpenTelemetryAttributes.PULSAR_PRODUCER_ID, 0) + .put(OpenTelemetryAttributes.PULSAR_PRODUCER_ACCESS_MODE, "shared") + .put(OpenTelemetryAttributes.PULSAR_DOMAIN, "non-persistent") + .put(OpenTelemetryAttributes.PULSAR_TENANT, "my-property") + .put(OpenTelemetryAttributes.PULSAR_NAMESPACE, "my-property/my-ns") + .put(OpenTelemetryAttributes.PULSAR_TOPIC, topic) + .build(); + var metrics = pulsarTestContext.getOpenTelemetryMetricReader().collectAllMetrics(); + assertMetricLongSumValue(metrics, OpenTelemetryProducerStats.MESSAGE_DROP_COUNTER, attributes, + value -> assertThat(value).isPositive()); } finally { conf.setMaxConcurrentNonPersistentMessagePerConnection(defaultNonPersistentMessageRate); } From edb4d63e906d481a0bee20ded75037d9d753e364 Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Fri, 7 Jun 2024 12:55:40 -0700 Subject: [PATCH 18/21] Fix BrokerServiceLookupTest.testMultipleBrokerLookup --- .../org/apache/pulsar/client/api/BrokerServiceLookupTest.java | 1 + 1 file changed, 1 insertion(+) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java index 336728f279eda..e99802a5bc5c4 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java @@ -192,6 +192,7 @@ public void testMultipleBrokerLookup() throws Exception { // Disable collecting topic stats during this test, as it deadlocks on access to map BrokerService.topics. pulsar2.getOpenTelemetryTopicStats().close(); pulsar2.getOpenTelemetryConsumerStats().close(); + pulsar2.getOpenTelemetryProducerStats().close(); var metricReader = pulsarTestContext.getOpenTelemetryMetricReader(); var lookupRequestSemaphoreField = BrokerService.class.getDeclaredField("lookupRequestSemaphore"); From 88a40bea560d6169142db1645fe6ab405d967489 Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Fri, 7 Jun 2024 15:26:17 -0700 Subject: [PATCH 19/21] Do not expose new fields in JSON --- .../policies/data/stats/NonPersistentPublisherStatsImpl.java | 1 + .../pulsar/common/policies/data/stats/PublisherStatsImpl.java | 3 +++ 2 files changed, 4 insertions(+) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/NonPersistentPublisherStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/NonPersistentPublisherStatsImpl.java index 8f7fc01e64b7f..d62e9b8dbbeae 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/NonPersistentPublisherStatsImpl.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/NonPersistentPublisherStatsImpl.java @@ -57,6 +57,7 @@ public void recordMsgDrop(long numMessages) { msgDrop.recordEvent(numMessages); } + @JsonIgnore public long getMsgDropCount() { return msgDrop.getTotalCount(); } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java index 9326311d67cc2..3f9067eba0b25 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java @@ -128,10 +128,12 @@ public void recordMsgIn(long messageCount, long byteCount) { msgIn.recordMultipleEvents(messageCount, byteCount); } + @JsonIgnore public long getMsgInCounter() { return msgIn.getTotalCount(); } + @JsonIgnore public long getBytesInCounter() { return msgIn.getTotalValue(); } @@ -140,6 +142,7 @@ public void recordChunkedMsgIn() { msgChunkIn.recordEvent(); } + @JsonIgnore public long getChunkedMsgInCounter() { return msgChunkIn.getTotalCount(); } From c6577c81bbb951589a969c818b0576be39f158ad Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Fri, 7 Jun 2024 16:38:55 -0700 Subject: [PATCH 20/21] Use lower_underscore case format for access mode enum attribute --- .../main/java/org/apache/pulsar/broker/service/Producer.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java index 3d31f4e00fb45..47c6dc8571d2d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java @@ -23,6 +23,7 @@ import static org.apache.pulsar.common.protocol.Commands.hasChecksum; import static org.apache.pulsar.common.protocol.Commands.readChecksum; import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.CaseFormat; import com.google.common.base.MoreObjects; import io.netty.buffer.ByteBuf; import io.netty.util.Recycler; @@ -871,7 +872,8 @@ public Attributes getOpenTelemetryAttributes() { var builder = Attributes.builder() .put(OpenTelemetryAttributes.PULSAR_PRODUCER_NAME, producerName) .put(OpenTelemetryAttributes.PULSAR_PRODUCER_ID, producerId) - .put(OpenTelemetryAttributes.PULSAR_PRODUCER_ACCESS_MODE, accessMode.name().toLowerCase()) + .put(OpenTelemetryAttributes.PULSAR_PRODUCER_ACCESS_MODE, + CaseFormat.UPPER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, accessMode.name())) .put(OpenTelemetryAttributes.PULSAR_DOMAIN, topicName.getDomain().toString()) .put(OpenTelemetryAttributes.PULSAR_TENANT, topicName.getTenant()) .put(OpenTelemetryAttributes.PULSAR_NAMESPACE, topicName.getNamespace()) From c24d7e8416e13598c03ae968561abb4a544ce2d4 Mon Sep 17 00:00:00 2001 From: Dragos Misca Date: Wed, 12 Jun 2024 13:25:17 -0700 Subject: [PATCH 21/21] Update OpenTelemetryProducerStats.java --- .../apache/pulsar/broker/stats/OpenTelemetryProducerStats.java | 1 - 1 file changed, 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java index 78a8067743644..9c09804554c31 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryProducerStats.java @@ -36,7 +36,6 @@ public class OpenTelemetryProducerStats implements AutoCloseable { public static final String BYTES_IN_COUNTER = "pulsar.broker.producer.message.incoming.size"; private final ObservableLongMeasurement bytesInCounter; - // Replaces pulsar_consumer_msg_ack_rate public static final String MESSAGE_DROP_COUNTER = "pulsar.broker.producer.message.drop.count"; private final ObservableLongMeasurement messageDropCounter;