[improve][broker] Reuse topic OpenTelemetry attributes

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PersistentTopicAttributes.java

@@ -0,0 +1,73 @@
+/*
+ * 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.service;
+
+import io.opentelemetry.api.common.Attributes;
+import lombok.Getter;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes;
+
+@Getter
+public class PersistentTopicAttributes extends TopicAttributes {
+
+    private final Attributes timeBasedQuotaAttributes;
+    private final Attributes sizeBasedQuotaAttributes;
+
+    private final Attributes compactionSuccessAttributes;
+    private final Attributes compactionFailureAttributes;
+
+    private final Attributes transactionActiveAttributes;
+    private final Attributes transactionCommittedAttributes;
+    private final Attributes transactionAbortedAttributes;
+
+    public PersistentTopicAttributes(TopicName topicName) {
+        super(topicName);
+
+        timeBasedQuotaAttributes = Attributes.builder()
+                .putAll(commonAttributes)
+                .putAll(OpenTelemetryAttributes.BacklogQuotaType.TIME.attributes)
+                .build();
+        sizeBasedQuotaAttributes = Attributes.builder()
+                .putAll(commonAttributes)
+                .putAll(OpenTelemetryAttributes.BacklogQuotaType.SIZE.attributes)
+                .build();
+
+        transactionActiveAttributes =  Attributes.builder()
+                .putAll(commonAttributes)
+                .putAll(OpenTelemetryAttributes.TransactionStatus.ACTIVE.attributes)
+                .build();
+        transactionCommittedAttributes =  Attributes.builder()
+                .putAll(commonAttributes)
+                .putAll(OpenTelemetryAttributes.TransactionStatus.COMMITTED.attributes)
+                .build();
+        transactionAbortedAttributes =  Attributes.builder()
+                .putAll(commonAttributes)
+                .putAll(OpenTelemetryAttributes.TransactionStatus.ABORTED.attributes)
+                .build();
+
+        compactionSuccessAttributes = Attributes.builder()
+                .putAll(commonAttributes)
+                .putAll(OpenTelemetryAttributes.CompactionStatus.SUCCESS.attributes)
+                .build();
+        compactionFailureAttributes = Attributes.builder()
+                .putAll(commonAttributes)
+                .putAll(OpenTelemetryAttributes.CompactionStatus.FAILURE.attributes)
+                .build();
+    }
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java

@@ -384,4 +384,9 @@ default boolean isSystemTopic() {
      */
     HierarchyTopicPolicies getHierarchyTopicPolicies();
 
+    /**
+     * Get OpenTelemetry attribute set.
+     * @return
+     */
+    TopicAttributes getTopicAttributes();
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicAttributes.java

@@ -0,0 +1,44 @@
+/*
+ * 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.service;
+
+import io.opentelemetry.api.common.Attributes;
+import java.util.Objects;
+import lombok.Getter;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes;
+
+@Getter
+public class TopicAttributes {
+
+    protected final Attributes commonAttributes;
+
+    public TopicAttributes(TopicName topicName) {
+        Objects.requireNonNull(topicName);
+        var builder = Attributes.builder()
+                .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());
+        }
+        commonAttributes = builder.build();
+    }
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java

@@ -36,6 +36,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import org.apache.bookkeeper.mledger.Entry;
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.commons.lang3.mutable.MutableInt;
@@ -62,6 +63,7 @@
 import org.apache.pulsar.broker.service.Subscription;
 import org.apache.pulsar.broker.service.SubscriptionOption;
 import org.apache.pulsar.broker.service.Topic;
+import org.apache.pulsar.broker.service.TopicAttributes;
 import org.apache.pulsar.broker.service.TopicPolicyListener;
 import org.apache.pulsar.broker.service.TransportCnx;
 import org.apache.pulsar.broker.stats.ClusterReplicationMetrics;
@@ -117,6 +119,11 @@ protected TopicStats initialValue() {
         }
     };
 
+    private volatile TopicAttributes topicAttributes = null;
+    private static final AtomicReferenceFieldUpdater<NonPersistentTopic, TopicAttributes>
+            TOPIC_ATTRIBUTES_FIELD_UPDATER = AtomicReferenceFieldUpdater.newUpdater(
+                    NonPersistentTopic.class, TopicAttributes.class, "topicAttributes");
+
     private static class TopicStats {
         public double averageMsgSize;
         public double aggMsgRateIn;
@@ -1268,4 +1275,13 @@ public boolean isPersistent() {
     public long getBestEffortOldestUnacknowledgedMessageAgeSeconds() {
         return -1;
     }
+
+    @Override
+    public TopicAttributes getTopicAttributes() {
+        if (topicAttributes != null) {
+            return topicAttributes;
+        }
+        return TOPIC_ATTRIBUTES_FIELD_UPDATER.updateAndGet(this,
+                old -> old != null ? old : new TopicAttributes(TopicName.get(topic)));
+    }
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -123,6 +123,7 @@
 import org.apache.pulsar.broker.service.Consumer;
 import org.apache.pulsar.broker.service.Dispatcher;
 import org.apache.pulsar.broker.service.GetStatsOptions;
+import org.apache.pulsar.broker.service.PersistentTopicAttributes;
 import org.apache.pulsar.broker.service.Producer;
 import org.apache.pulsar.broker.service.Replicator;
 import org.apache.pulsar.broker.service.StreamingStats;
@@ -291,6 +292,11 @@ protected TopicStatsHelper initialValue() {
     @Getter
     private final PersistentTopicMetrics persistentTopicMetrics = new PersistentTopicMetrics();
 
+    private volatile PersistentTopicAttributes persistentTopicAttributes = null;
+    private static final AtomicReferenceFieldUpdater<PersistentTopic, PersistentTopicAttributes>
+            PERSISTENT_TOPIC_ATTRIBUTES_FIELD_UPDATER = AtomicReferenceFieldUpdater.newUpdater(
+                    PersistentTopic.class, PersistentTopicAttributes.class, "persistentTopicAttributes");
+
     private volatile TimeBasedBacklogQuotaCheckResult timeBasedBacklogQuotaCheckResult;
     private static final AtomicReferenceFieldUpdater<PersistentTopic, TimeBasedBacklogQuotaCheckResult>
             TIME_BASED_BACKLOG_QUOTA_CHECK_RESULT_UPDATER = AtomicReferenceFieldUpdater.newUpdater(
@@ -4355,4 +4361,12 @@ protected boolean isExceedMaximumDeliveryDelay(ByteBuf headersAndPayload) {
         return false;
     }
 
+    @Override
+    public PersistentTopicAttributes getTopicAttributes() {
+        if (persistentTopicAttributes != null) {
+            return persistentTopicAttributes;
+        }
+        return PERSISTENT_TOPIC_ATTRIBUTES_FIELD_UPDATER.updateAndGet(this,
+                old -> old != null ? old : new PersistentTopicAttributes(TopicName.get(topic)));
+    }
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/OpenTelemetryTopicStats.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.ObservableDoubleMeasurement;
 import io.opentelemetry.api.metrics.ObservableLongMeasurement;
@@ -31,12 +30,10 @@
 import org.apache.pulsar.broker.service.Subscription;
 import org.apache.pulsar.broker.service.Topic;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
-import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.policies.data.BacklogQuota;
 import org.apache.pulsar.common.stats.MetricsUtil;
 import org.apache.pulsar.compaction.CompactedTopicContext;
 import org.apache.pulsar.compaction.Compactor;
-import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes;
 
 public class OpenTelemetryTopicStats implements AutoCloseable {
 
@@ -383,16 +380,8 @@ public void close() {
     }
 
     private void recordMetricsForTopic(Topic topic) {
-        var topicName = TopicName.get(topic.getName());
-        var builder = Attributes.builder()
-                .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 topicAttributes = topic.getTopicAttributes();
+        var attributes = topicAttributes.getCommonAttributes();
 
         if (topic instanceof AbstractTopic abstractTopic) {
             subscriptionCounter.record(abstractTopic.getSubscriptions().size(), attributes);
@@ -410,6 +399,7 @@ private void recordMetricsForTopic(Topic topic) {
         }
 
         if (topic instanceof PersistentTopic persistentTopic) {
+            var persistentTopicAttributes = persistentTopic.getTopicAttributes();
             var managedLedger = persistentTopic.getManagedLedger();
             var managedLedgerStats = persistentTopic.getManagedLedger().getStats();
             storageCounter.record(managedLedgerStats.getStoredMessagesSize(), attributes);
@@ -428,45 +418,27 @@ private void recordMetricsForTopic(Topic topic) {
             backlogQuotaAge.record(topic.getBestEffortOldestUnacknowledgedMessageAgeSeconds(), attributes);
             var backlogQuotaMetrics = persistentTopic.getPersistentTopicMetrics().getBacklogQuotaMetrics();
             backlogEvictionCounter.record(backlogQuotaMetrics.getSizeBasedBacklogQuotaExceededEvictionCount(),
-                    Attributes.builder()
-                            .putAll(attributes)
-                            .put(OpenTelemetryAttributes.PULSAR_BACKLOG_QUOTA_TYPE, "size")
-                            .build());
+                    persistentTopicAttributes.getSizeBasedQuotaAttributes());
             backlogEvictionCounter.record(backlogQuotaMetrics.getTimeBasedBacklogQuotaExceededEvictionCount(),
-                    Attributes.builder()
-                            .putAll(attributes)
-                            .put(OpenTelemetryAttributes.PULSAR_BACKLOG_QUOTA_TYPE, "time")
-                            .build());
+                    persistentTopicAttributes.getTimeBasedQuotaAttributes());
 
             var txnBuffer = persistentTopic.getTransactionBuffer();
-            transactionCounter.record(txnBuffer.getOngoingTxnCount(), Attributes.builder()
-                    .putAll(attributes)
-                    .put(OpenTelemetryAttributes.PULSAR_TRANSACTION_STATUS, "active")
-                    .build());
-            transactionCounter.record(txnBuffer.getCommittedTxnCount(), Attributes.builder()
-                    .putAll(attributes)
-                    .put(OpenTelemetryAttributes.PULSAR_TRANSACTION_STATUS, "committed")
-                    .build());
-            transactionCounter.record(txnBuffer.getAbortedTxnCount(), Attributes.builder()
-                    .putAll(attributes)
-                    .put(OpenTelemetryAttributes.PULSAR_TRANSACTION_STATUS, "aborted")
-                    .build());
+            transactionCounter.record(txnBuffer.getOngoingTxnCount(),
+                    persistentTopicAttributes.getTransactionActiveAttributes());
+            transactionCounter.record(txnBuffer.getCommittedTxnCount(),
+                    persistentTopicAttributes.getTransactionCommittedAttributes());
+            transactionCounter.record(txnBuffer.getAbortedTxnCount(),
+                    persistentTopicAttributes.getTransactionAbortedAttributes());
 
             Optional.ofNullable(pulsar.getNullableCompactor())
                     .map(Compactor::getStats)
                     .flatMap(compactorMXBean -> compactorMXBean.getCompactionRecordForTopic(topic.getName()))
                     .ifPresent(compactionRecord -> {
                         compactionRemovedCounter.record(compactionRecord.getCompactionRemovedEventCount(), attributes);
                         compactionOperationCounter.record(compactionRecord.getCompactionSucceedCount(),
-                                Attributes.builder()
-                                        .putAll(attributes)
-                                        .put(OpenTelemetryAttributes.PULSAR_COMPACTION_STATUS, "success")
-                                        .build());
+                                persistentTopicAttributes.getCompactionSuccessAttributes());
                         compactionOperationCounter.record(compactionRecord.getCompactionFailedCount(),
-                                Attributes.builder()
-                                        .putAll(attributes)
-                                        .put(OpenTelemetryAttributes.PULSAR_COMPACTION_STATUS, "failure")
-                                        .build());
+                                persistentTopicAttributes.getCompactionFailureAttributes());
                         compactionDurationSeconds.record(MetricsUtil.convertToSeconds(
                             compactionRecord.getCompactionDurationTimeInMills(), TimeUnit.MILLISECONDS), attributes);
                         compactionBytesInCounter.record(compactionRecord.getCompactionReadBytes(), attributes);

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 io.opentelemetry.api.common.Attributes;
 import java.util.List;
 
 /**
@@ -100,14 +101,30 @@ public interface OpenTelemetryAttributes {
      * The status of the Pulsar transaction.
      */
     AttributeKey<String> PULSAR_TRANSACTION_STATUS = AttributeKey.stringKey("pulsar.transaction.status");
+    enum TransactionStatus {
+        ACTIVE,
+        COMMITTED,
+        ABORTED;
+        public final Attributes attributes = Attributes.of(PULSAR_TRANSACTION_STATUS, name().toLowerCase());
+    }
 
     /**
      * The status of the Pulsar compaction operation.
      */
     AttributeKey<String> PULSAR_COMPACTION_STATUS = AttributeKey.stringKey("pulsar.compaction.status");
+    enum CompactionStatus {
+        SUCCESS,
+        FAILURE;
+        public final Attributes attributes = Attributes.of(PULSAR_COMPACTION_STATUS, name().toLowerCase());
+    }
 
     /**
      * The type of the backlog quota.
      */
     AttributeKey<String> PULSAR_BACKLOG_QUOTA_TYPE = AttributeKey.stringKey("pulsar.backlog.quota.type");
+    enum BacklogQuotaType {
+        SIZE,
+        TIME;
+        public final Attributes attributes = Attributes.of(PULSAR_BACKLOG_QUOTA_TYPE, name().toLowerCase());
+    }
 }