[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);
+ }
+}