[feat][broker] PIP-264: Add OpenTelemetry managed cursor metrics

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java

@@ -877,4 +877,12 @@ default boolean periodicRollover() {
         return false;
     }
 
+    /**
+     * Get the attributes associated with the cursor.
+     *
+     * @return the attributes associated with the cursor
+     */
+    default ManagedCursorAttributes getManagedCursorAttributes() {
+        return new ManagedCursorAttributes(this);
+    }
 }

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursorAttributes.java

@@ -0,0 +1,51 @@
+/*
+ * 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.bookkeeper.mledger;
+
+import io.opentelemetry.api.common.Attributes;
+import lombok.Getter;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes;
+import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes.ManagedCursorOperationStatus;
+
+@Getter
+public class ManagedCursorAttributes {
+
+    private final Attributes attributes;
+    private final Attributes attributesOperationSucceed;
+    private final Attributes attributesOperationFailure;
+
+    public ManagedCursorAttributes(ManagedCursor cursor) {
+        var mlName = cursor.getManagedLedger().getName();
+        var topicName = TopicName.get(TopicName.fromPersistenceNamingEncoding(mlName));
+        attributes = Attributes.of(
+                OpenTelemetryAttributes.ML_CURSOR_NAME, cursor.getName(),
+                OpenTelemetryAttributes.ML_LEDGER_NAME, mlName,
+                OpenTelemetryAttributes.PULSAR_NAMESPACE, topicName.getNamespace()
+        );
+        attributesOperationSucceed = Attributes.builder()
+                .putAll(attributes)
+                .putAll(ManagedCursorOperationStatus.SUCCESS.attributes)
+                .build();
+        attributesOperationFailure = Attributes.builder()
+                .putAll(attributes)
+                .putAll(ManagedCursorOperationStatus.FAILURE.attributes)
+                .build();
+    }
+}

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

@@ -77,6 +77,7 @@
 import org.apache.bookkeeper.mledger.AsyncCallbacks.SkipEntriesCallback;
 import org.apache.bookkeeper.mledger.Entry;
 import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.ManagedCursorAttributes;
 import org.apache.bookkeeper.mledger.ManagedCursorMXBean;
 import org.apache.bookkeeper.mledger.ManagedLedger;
 import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
@@ -286,6 +287,11 @@ public enum State {
 
     protected final ManagedCursorMXBean mbean;
 
+    private volatile ManagedCursorAttributes managedCursorAttributes;
+    private static final AtomicReferenceFieldUpdater<ManagedCursorImpl, ManagedCursorAttributes> ATTRIBUTES_UPDATER =
+            AtomicReferenceFieldUpdater.newUpdater(ManagedCursorImpl.class, ManagedCursorAttributes.class,
+                    "managedCursorAttributes");
+
     @SuppressWarnings("checkstyle:javadoctype")
     public interface VoidCallback {
         void operationComplete();
@@ -3719,4 +3725,12 @@ public ManagedCursor duplicateNonDurableCursor(String nonDurableCursorName) thro
         }
         return newNonDurableCursor;
     }
+
+    @Override
+    public ManagedCursorAttributes getManagedCursorAttributes() {
+        if (managedCursorAttributes != null) {
+            return managedCursorAttributes;
+        }
+        return ATTRIBUTES_UPDATER.updateAndGet(this, old -> old != null ? old : new ManagedCursorAttributes(this));
+    }
 }

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java

@@ -122,6 +122,7 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
 
     private final OpenTelemetryManagedLedgerCacheStats openTelemetryCacheStats;
     private final OpenTelemetryManagedLedgerStats openTelemetryManagedLedgerStats;
+    private final OpenTelemetryManagedCursorStats openTelemetryManagedCursorStats;
 
     //indicate whether shutdown() is called.
     private volatile boolean closed;
@@ -231,6 +232,7 @@ private ManagedLedgerFactoryImpl(MetadataStoreExtended metadataStore,
 
         openTelemetryCacheStats = new OpenTelemetryManagedLedgerCacheStats(openTelemetry, this);
         openTelemetryManagedLedgerStats = new OpenTelemetryManagedLedgerStats(openTelemetry, this);
+        openTelemetryManagedCursorStats = new OpenTelemetryManagedCursorStats(openTelemetry, this);
     }
 
     static class DefaultBkFactory implements BookkeeperFactoryForCustomEnsemblePlacementPolicy {
@@ -622,6 +624,7 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) {
                     }));
                 }).thenAcceptAsync(__ -> {
                     //wait for tasks in scheduledExecutor executed.
+                    openTelemetryManagedCursorStats.close();
                     openTelemetryManagedLedgerStats.close();
                     openTelemetryCacheStats.close();
                     scheduledExecutor.shutdownNow();

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpenTelemetryManagedCursorStats.java

@@ -0,0 +1,136 @@
+/*
+ * 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.bookkeeper.mledger.impl;
+
+import com.google.common.collect.Streams;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.metrics.BatchCallback;
+import io.opentelemetry.api.metrics.ObservableLongMeasurement;
+import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.pulsar.opentelemetry.Constants;
+
+public class OpenTelemetryManagedCursorStats implements AutoCloseable {
+
+    // Replaces ['pulsar_ml_cursor_persistLedgerSucceed', 'pulsar_ml_cursor_persistLedgerErrors']
+    public static final String PERSIST_OPERATION_COUNTER = "pulsar.broker.managed_ledger.persist.operation.count";
+    private final ObservableLongMeasurement persistOperationCounter;
+
+    // Replaces ['pulsar_ml_cursor_persistZookeeperSucceed', 'pulsar_ml_cursor_persistZookeeperErrors']
+    public static final String PERSIST_OPERATION_METADATA_STORE_COUNTER =
+            "pulsar.broker.managed_ledger.persist.mds.operation.count";
+    private final ObservableLongMeasurement persistOperationMetadataStoreCounter;
+
+    // Replaces pulsar_ml_cursor_nonContiguousDeletedMessagesRange
+    public static final String NON_CONTIGUOUS_MESSAGE_RANGE_COUNTER =
+            "pulsar.broker.managed_ledger.message_range.count";
+    private final ObservableLongMeasurement nonContiguousMessageRangeCounter;
+
+    // Replaces pulsar_ml_cursor_writeLedgerSize
+    public static final String OUTGOING_BYTE_COUNTER = "pulsar.broker.managed_ledger.cursor.outgoing.size";
+    private final ObservableLongMeasurement outgoingByteCounter;
+
+    // Replaces pulsar_ml_cursor_writeLedgerLogicalSize
+    public static final String OUTGOING_BYTE_LOGICAL_COUNTER =
+            "pulsar.broker.managed_ledger.cursor.outgoing.logical.size";
+    private final ObservableLongMeasurement outgoingByteLogicalCounter;
+
+    // Replaces pulsar_ml_cursor_readLedgerSize
+    public static final String INCOMING_BYTE_COUNTER = "pulsar.broker.managed_ledger.cursor.incoming.size";
+    private final ObservableLongMeasurement incomingByteCounter;
+
+    private final BatchCallback batchCallback;
+
+    public OpenTelemetryManagedCursorStats(OpenTelemetry openTelemetry, ManagedLedgerFactoryImpl factory) {
+        var meter = openTelemetry.getMeter(Constants.BROKER_INSTRUMENTATION_SCOPE_NAME);
+
+        persistOperationCounter = meter
+                .counterBuilder(PERSIST_OPERATION_COUNTER)
+                .setUnit("{operation}")
+                .setDescription("The number of acknowledgment operations on the ledger.")
+                .buildObserver();
+
+        persistOperationMetadataStoreCounter = meter
+                .counterBuilder(PERSIST_OPERATION_METADATA_STORE_COUNTER)
+                .setUnit("{operation}")
+                .setDescription("The number of acknowledgment operations in the metadata store.")
+                .buildObserver();
+
+        nonContiguousMessageRangeCounter = meter
+                .upDownCounterBuilder(NON_CONTIGUOUS_MESSAGE_RANGE_COUNTER)
+                .setUnit("{range}")
+                .setDescription("The number of non-contiguous deleted messages ranges.")
+                .buildObserver();
+
+        outgoingByteCounter = meter
+                .counterBuilder(OUTGOING_BYTE_COUNTER)
+                .setUnit("{By}")
+                .setDescription("The total amount of data written to the ledger.")
+                .buildObserver();
+
+        outgoingByteLogicalCounter = meter
+                .counterBuilder(OUTGOING_BYTE_LOGICAL_COUNTER)
+                .setUnit("{By}")
+                .setDescription("The total amount of data written to the ledger, not including replicas.")
+                .buildObserver();
+
+        incomingByteCounter = meter
+                .counterBuilder(INCOMING_BYTE_COUNTER)
+                .setUnit("{By}")
+                .setDescription("The total amount of data read from the ledger.")
+                .buildObserver();
+
+        batchCallback = meter.batchCallback(() -> factory.getManagedLedgers()
+                        .values()
+                        .stream()
+                        .map(ManagedLedgerImpl::getCursors)
+                        .flatMap(Streams::stream)
+                        .forEach(this::recordMetrics),
+                persistOperationCounter,
+                persistOperationMetadataStoreCounter,
+                nonContiguousMessageRangeCounter,
+                outgoingByteCounter,
+                outgoingByteLogicalCounter,
+                incomingByteCounter);
+    }
+
+    @Override
+    public void close() {
+        batchCallback.close();
+    }
+
+    private void recordMetrics(ManagedCursor cursor) {
+        var stats = cursor.getStats();
+        var cursorAttributesSet = cursor.getManagedCursorAttributes();
+        var attributes = cursorAttributesSet.getAttributes();
+        var attributesSucceed = cursorAttributesSet.getAttributesOperationSucceed();
+        var attributesFailed = cursorAttributesSet.getAttributesOperationFailure();
+
+        persistOperationCounter.record(stats.getPersistLedgerSucceed(), attributesSucceed);
+        persistOperationCounter.record(stats.getPersistLedgerErrors(), attributesFailed);
+
+        persistOperationMetadataStoreCounter.record(stats.getPersistZookeeperSucceed(), attributesSucceed);
+        persistOperationMetadataStoreCounter.record(stats.getPersistZookeeperErrors(), attributesFailed);
+
+        nonContiguousMessageRangeCounter.record(cursor.getTotalNonContiguousDeletedMessagesRange(), attributes);
+
+        outgoingByteCounter.record(stats.getWriteCursorLedgerSize(), attributes);
+        outgoingByteLogicalCounter.record(stats.getWriteCursorLedgerLogicalSize(), attributes);
+        incomingByteCounter.record(stats.getReadCursorLedgerSize(), attributes);
+    }
+}