[improve][client] Extract OTel metrics into ProducerMetrics/ConsumerMetrics classes

Author: hqbfzwang

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

@@ -34,7 +34,7 @@
 import io.netty.util.ReferenceCountUtil;
 import io.netty.util.Timeout;
 import io.netty.util.concurrent.FastThreadLocal;
-import io.opentelemetry.api.common.Attributes;
+
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -98,10 +98,8 @@
 import org.apache.pulsar.client.api.transaction.TxnID;
 import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
 import org.apache.pulsar.client.impl.crypto.MessageCryptoBc;
-import org.apache.pulsar.client.impl.metrics.Counter;
+import org.apache.pulsar.client.impl.metrics.ConsumerMetrics;
 import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
-import org.apache.pulsar.client.impl.metrics.Unit;
-import org.apache.pulsar.client.impl.metrics.UpDownCounter;
 import org.apache.pulsar.client.impl.schema.AutoConsumeSchema;
 import org.apache.pulsar.client.impl.schema.AutoProduceBytesSchema;
 import org.apache.pulsar.client.impl.transaction.TransactionImpl;
@@ -228,16 +226,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
     private final boolean createTopicIfDoesNotExist;
     private final boolean poolMessages;
 
-    private final Counter messagesReceivedCounter;
-    private final Counter bytesReceivedCounter;
-    private final UpDownCounter messagesPrefetchedGauge;
-    private final UpDownCounter bytesPrefetchedGauge;
-    private final Counter consumersOpenedCounter;
-    private final Counter consumersClosedCounter;
-    private final Counter consumerAcksCounter;
-    private final Counter consumerNacksCounter;
-
-    private final Counter consumerDlqMessagesCounter;
+    private final ConsumerMetrics consumerMetrics;
 
     private final AtomicReference<ClientCnx> clientCnxUsedForConsumerRegistration = new AtomicReference<>();
     private final AtomicInteger previousExceptionCount = new AtomicInteger();
@@ -421,29 +410,10 @@ protected ConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurat
         topicNameWithoutPartition = topicName.getPartitionedTopicName();
 
         InstrumentProvider ip = client.instrumentProvider();
-        Attributes attrs = Attributes.builder().put("pulsar.subscription", subscription).build();
-        consumersOpenedCounter = ip.newCounter("pulsar.client.consumer.opened", Unit.Sessions,
-                "The number of consumer sessions opened", topic, attrs);
-        consumersClosedCounter = ip.newCounter("pulsar.client.consumer.closed", Unit.Sessions,
-                "The number of consumer sessions closed", topic, attrs);
-        messagesReceivedCounter = ip.newCounter("pulsar.client.consumer.message.received.count", Unit.Messages,
-                "The number of messages explicitly received by the consumer application", topic, attrs);
-        bytesReceivedCounter = ip.newCounter("pulsar.client.consumer.message.received.size", Unit.Bytes,
-                "The number of bytes explicitly received by the consumer application", topic, attrs);
-        messagesPrefetchedGauge = ip.newUpDownCounter("pulsar.client.consumer.receive_queue.count", Unit.Messages,
-                "The number of messages currently sitting in the consumer receive queue", topic, attrs);
-        bytesPrefetchedGauge = ip.newUpDownCounter("pulsar.client.consumer.receive_queue.size", Unit.Bytes,
-                "The total size in bytes of messages currently sitting in the consumer receive queue", topic, attrs);
-
-        consumerAcksCounter = ip.newCounter("pulsar.client.consumer.message.ack", Unit.Messages,
-                "The number of acknowledged messages", topic, attrs);
-        consumerNacksCounter = ip.newCounter("pulsar.client.consumer.message.nack", Unit.Messages,
-                "The number of negatively acknowledged messages", topic, attrs);
-        consumerDlqMessagesCounter = ip.newCounter("pulsar.client.consumer.message.dlq", Unit.Messages,
-                "The number of messages sent to DLQ", topic, attrs);
+        consumerMetrics = new ConsumerMetrics(ip, topic, subscription);
         grabCnx();
 
-        consumersOpenedCounter.increment();
+        consumerMetrics.recordConsumerOpened();
     }
 
     public ConnectionHandler getConnectionHandler() {
@@ -607,7 +577,7 @@ protected CompletableFuture<Messages<T>> internalBatchReceiveAsync() {
     protected CompletableFuture<Void> doAcknowledge(MessageId messageId, AckType ackType,
                                                     Map<String, Long> properties,
                                                     TransactionImpl txn) {
-        consumerAcksCounter.increment();
+        consumerMetrics.recordAck();
 
         if (getState() != State.Ready && getState() != State.Connecting) {
             stats.incrementNumAcksFailed();
@@ -630,7 +600,7 @@ protected CompletableFuture<Void> doAcknowledge(MessageId messageId, AckType ack
     @Override
     protected CompletableFuture<Void> doAcknowledge(List<MessageId> messageIdList, AckType ackType,
                                                     Map<String, Long> properties, TransactionImpl txn) {
-        consumerAcksCounter.increment();
+        consumerMetrics.recordAck();
 
         if (getState() != State.Ready && getState() != State.Connecting) {
             stats.incrementNumAcksFailed();
@@ -721,7 +691,7 @@ protected CompletableFuture<Void> doReconsumeLater(Message<?> message, AckType a
                             copyMessageKeysIfNeeded(message, typedMessageBuilderNew);
                             copyMessageEventTime(message, typedMessageBuilderNew);
                             typedMessageBuilderNew.sendAsync().thenAccept(msgId -> {
-                                consumerDlqMessagesCounter.increment();
+                                consumerMetrics.recordDlq();
 
                                 doAcknowledge(finalMessageId, ackType, Collections.emptyMap(), null).thenAccept(v -> {
                                     result.complete(null);
@@ -835,7 +805,7 @@ private static void copyMessageEventTime(Message<?> message,
 
     @Override
     public void negativeAcknowledge(MessageId messageId) {
-        consumerNacksCounter.increment();
+        consumerMetrics.recordNack();
         negativeAcksTracker.add(messageId);
 
         // Ensure the message is not redelivered for ack-timeout, since we did receive an "ack"
@@ -844,7 +814,7 @@ public void negativeAcknowledge(MessageId messageId) {
 
     @Override
     public void negativeAcknowledge(Message<?> message) {
-        consumerNacksCounter.increment();
+        consumerMetrics.recordNack();
         negativeAcksTracker.add(message);
 
         // Ensure the message is not redelivered for ack-timeout, since we did receive an "ack"
@@ -1185,7 +1155,7 @@ public synchronized CompletableFuture<Void> closeAsync() {
             return compositeCloseFuture;
         }
 
-        consumersClosedCounter.increment();
+        consumerMetrics.recordConsumerClosed();
 
         if (!isConnected()) {
             log.info("[{}] [{}] Closed Consumer (not connected)", topic, subscription);
@@ -1369,8 +1339,7 @@ protected <V> MessageImpl<V> newMessage(final MessageIdImpl messageId,
     }
 
     private void executeNotifyCallback(final MessageImpl<T> message) {
-        messagesPrefetchedGauge.increment();
-        bytesPrefetchedGauge.add(message.size());
+        consumerMetrics.recordMessagePrefetched(message.size());
 
         // Enqueue the message so that it can be retrieved when application calls receive()
         // if the conf.getReceiverQueueSize() is 0 then discard message if no one is waiting for it.
@@ -1869,11 +1838,7 @@ protected synchronized void messageProcessed(Message<?> msg) {
         ClientCnx msgCnx = ((MessageImpl<?>) msg).getCnx();
         lastDequeuedMessageId = msg.getMessageId();
 
-        messagesPrefetchedGauge.decrement();
-        messagesReceivedCounter.increment();
-
-        bytesPrefetchedGauge.subtract(msg.size());
-        bytesReceivedCounter.add(msg.size());
+        consumerMetrics.recordMessageReceived(msg.size());
 
         if (msgCnx != currentCnx) {
             // The processed message did belong to the old queue that was cleared after reconnection.

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

@@ -81,11 +81,9 @@
 import org.apache.pulsar.client.api.transaction.Transaction;
 import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
 import org.apache.pulsar.client.impl.crypto.MessageCryptoBc;
-import org.apache.pulsar.client.impl.metrics.Counter;
 import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
 import org.apache.pulsar.client.impl.metrics.LatencyHistogram;
-import org.apache.pulsar.client.impl.metrics.Unit;
-import org.apache.pulsar.client.impl.metrics.UpDownCounter;
+import org.apache.pulsar.client.impl.metrics.ProducerMetrics;
 import org.apache.pulsar.client.impl.schema.JSONSchema;
 import org.apache.pulsar.client.impl.schema.SchemaUtils;
 import org.apache.pulsar.client.impl.transaction.TransactionImpl;
@@ -183,14 +181,9 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
 
     private boolean errorState;
 
-    private final LatencyHistogram latencyHistogram;
+    private final ProducerMetrics producerMetrics;
+    // rpcLatencyHistogram 需要传递给 OpSendMsg，保留包级别访问
     final LatencyHistogram rpcLatencyHistogram;
-    private final Counter publishedBytesCounter;
-    private final UpDownCounter pendingMessagesUpDownCounter;
-    private final UpDownCounter pendingBytesUpDownCounter;
-
-    private final Counter producersOpenedCounter;
-    private final Counter producersClosedCounter;
     private final boolean pauseSendingToPreservePublishOrderOnSchemaRegFailure;
     // This variable can be exposed as a metrics in the future, a PIP is needed.
     private final AtomicInteger pendingQueueFullCounter;
@@ -294,30 +287,14 @@ public ProducerImpl(PulsarClientImpl client, String topic, ProducerConfiguration
         }
 
         InstrumentProvider ip = client.instrumentProvider();
-        latencyHistogram = ip.newLatencyHistogram("pulsar.client.producer.message.send.duration",
-                "Publish latency experienced by the application, includes client batching time", topic,
-                Attributes.empty());
-        rpcLatencyHistogram = ip.newLatencyHistogram("pulsar.client.producer.rpc.send.duration",
-                "Publish RPC latency experienced internally by the client when sending data to receiving an ack", topic,
-                Attributes.empty());
-        publishedBytesCounter = ip.newCounter("pulsar.client.producer.message.send.size",
-                Unit.Bytes, "The number of bytes published", topic, Attributes.empty());
-        pendingMessagesUpDownCounter =
-                ip.newUpDownCounter("pulsar.client.producer.message.pending.count", Unit.Messages,
-                        "The number of messages in the producer internal send queue, waiting to be sent", topic,
-                        Attributes.empty());
-        pendingBytesUpDownCounter = ip.newUpDownCounter("pulsar.client.producer.message.pending.size", Unit.Bytes,
-                "The size of the messages in the producer internal queue, waiting to sent", topic, Attributes.empty());
-        producersOpenedCounter = ip.newCounter("pulsar.client.producer.opened", Unit.Sessions,
-                "The number of producer sessions opened", topic, Attributes.empty());
-        producersClosedCounter = ip.newCounter("pulsar.client.producer.closed", Unit.Sessions,
-                "The number of producer sessions closed", topic, Attributes.empty());
+        producerMetrics = new ProducerMetrics(ip, topic);
+        rpcLatencyHistogram = producerMetrics.rpcLatencyHistogram;
         pendingQueueFullCounter = new AtomicInteger();
 
         this.connectionHandler = initConnectionHandler();
         setChunkMaxMessageSize();
         grabCnx();
-        producersOpenedCounter.increment();
+        producerMetrics.recordProducerOpened();
     }
 
     @VisibleForTesting
@@ -398,8 +375,7 @@ CompletableFuture<MessageId> internalSendAsync(Message<?> message) {
         }
 
         int msgSize = interceptorMessage.getDataBuffer().readableBytes();
-        pendingMessagesUpDownCounter.increment();
-        pendingBytesUpDownCounter.add(msgSize);
+        producerMetrics.recordPendingMessage(msgSize);
 
         sendAsync(interceptorMessage, new DefaultSendMessageCallback(future, interceptorMessage, msgSize));
         return future;
@@ -457,22 +433,19 @@ private void onSendComplete(Throwable e, SendCallback sendCallback, MessageImpl<
             long createdAt = (sendCallback instanceof ProducerImpl.DefaultSendMessageCallback)
                     ? ((DefaultSendMessageCallback) sendCallback).createdAt : this.createdAt;
             long latencyNanos = System.nanoTime() - createdAt;
-            pendingMessagesUpDownCounter.decrement();
-            pendingBytesUpDownCounter.subtract(msgSize);
             ByteBuf payload = msg.getDataBuffer();
             if (payload == null) {
                 log.error("[{}] [{}] Payload is null when calling onSendComplete, which is not expected.",
                         topic, producerName);
             }
             try {
                 if (e != null) {
-                    latencyHistogram.recordFailure(latencyNanos);
+                    producerMetrics.recordSendFailed(latencyNanos, msgSize);
                     stats.incrementSendFailed();
                     onSendAcknowledgement(msg, null, e);
                     sendCallback.getFuture().completeExceptionally(e);
                 } else {
-                    latencyHistogram.recordSuccess(latencyNanos);
-                    publishedBytesCounter.add(msgSize);
+                    producerMetrics.recordSendSuccess(latencyNanos, msgSize);
                     stats.incrementNumAcksReceived(latencyNanos);
                     onSendAcknowledgement(msg, msg.getMessageId(), null);
                     sendCallback.getFuture().complete(msg.getMessageId());
@@ -1219,7 +1192,7 @@ public synchronized CompletableFuture<Void> closeAsync() {
             return CompletableFuture.completedFuture(null);
         }
 
-        producersClosedCounter.increment();
+        producerMetrics.recordProducerClosed();
         closeProducerTasks();
 
         ClientCnx cnx = cnx();

pulsar-client/src/main/java/org/apache/pulsar/client/impl/metrics/ConsumerMetrics.java

@@ -0,0 +1,89 @@
+/*
+ * 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.client.impl.metrics;
+
+import io.opentelemetry.api.common.Attributes;
+
+public class ConsumerMetrics {
+
+    private final Counter messagesReceivedCounter;
+    private final Counter bytesReceivedCounter;
+    private final UpDownCounter messagesPrefetchedGauge;
+    private final UpDownCounter bytesPrefetchedGauge;
+    private final Counter consumersOpenedCounter;
+    private final Counter consumersClosedCounter;
+    private final Counter consumerAcksCounter;
+    private final Counter consumerNacksCounter;
+    private final Counter consumerDlqMessagesCounter;
+
+    public ConsumerMetrics(InstrumentProvider ip, String topic, String subscription) {
+        Attributes attrs = Attributes.builder().put("pulsar.subscription", subscription).build();
+
+        consumersOpenedCounter = ip.newCounter("pulsar.client.consumer.opened", Unit.Sessions,
+                "The number of consumer sessions opened", topic, attrs);
+        consumersClosedCounter = ip.newCounter("pulsar.client.consumer.closed", Unit.Sessions,
+                "The number of consumer sessions closed", topic, attrs);
+        messagesReceivedCounter = ip.newCounter("pulsar.client.consumer.message.received.count", Unit.Messages,
+                "The number of messages explicitly received by the consumer application", topic, attrs);
+        bytesReceivedCounter = ip.newCounter("pulsar.client.consumer.message.received.size", Unit.Bytes,
+                "The number of bytes explicitly received by the consumer application", topic, attrs);
+        messagesPrefetchedGauge = ip.newUpDownCounter("pulsar.client.consumer.receive_queue.count", Unit.Messages,
+                "The number of messages currently sitting in the consumer receive queue", topic, attrs);
+        bytesPrefetchedGauge = ip.newUpDownCounter("pulsar.client.consumer.receive_queue.size", Unit.Bytes,
+                "The total size in bytes of messages currently sitting in the consumer receive queue", topic, attrs);
+        consumerAcksCounter = ip.newCounter("pulsar.client.consumer.message.ack", Unit.Messages,
+                "The number of acknowledged messages", topic, attrs);
+        consumerNacksCounter = ip.newCounter("pulsar.client.consumer.message.nack", Unit.Messages,
+                "The number of negatively acknowledged messages", topic, attrs);
+        consumerDlqMessagesCounter = ip.newCounter("pulsar.client.consumer.message.dlq", Unit.Messages,
+                "The number of messages sent to DLQ", topic, attrs);
+    }
+
+    public void recordMessagePrefetched(int msgSize) {
+        messagesPrefetchedGauge.increment();
+        bytesPrefetchedGauge.add(msgSize);
+    }
+
+    public void recordMessageReceived(int msgSize) {
+        messagesPrefetchedGauge.decrement();
+        bytesPrefetchedGauge.subtract(msgSize);
+        messagesReceivedCounter.increment();
+        bytesReceivedCounter.add(msgSize);
+    }
+
+    public void recordAck() {
+        consumerAcksCounter.increment();
+    }
+
+    public void recordNack() {
+        consumerNacksCounter.increment();
+    }
+
+    public void recordDlq() {
+        consumerDlqMessagesCounter.increment();
+    }
+
+    public void recordConsumerOpened() {
+        consumersOpenedCounter.increment();
+    }
+
+    public void recordConsumerClosed() {
+        consumersClosedCounter.increment();
+    }
+}