Add Cassandra compaction progress metrics via JMX handler

instrumentation/jmx-metrics/library/cassandra.md

@@ -2,15 +2,17 @@
 
 Here is the list of metrics based on MBeans exposed by Cassandra.
 
-| Metric Name                          | Type          | Unit      | Attributes                            | Description                                                      |
-| ------------------------------------ | ------------- | --------- | ------------------------------------- | ---------------------------------------------------------------- |
-| cassandra.client.request.count       | Counter       | {request} | cassandra.operation                   | Number of requests by operation.                                 |
-| cassandra.client.request.error       | Counter       | {error}   | cassandra.operation, cassandra.status | Number of request errors by operation.                           |
-| cassandra.client.request.latency.p50 | Gauge         | s         | cassandra.operation                   | Request latency 50th percentile by operation.                    |
-| cassandra.client.request.latency.p99 | Gauge         | s         | cassandra.operation                   | Request latency 99th percentile by operation.                    |
-| cassandra.client.request.latency.max | Gauge         | s         | cassandra.operation                   | Maximum request latency by operation.                            |
-| cassandra.compaction.tasks.completed | Counter       | {task}    |                                       | Number of completed compactions since server start.              |
-| cassandra.compaction.tasks.pending   | Gauge         | {task}    |                                       | Estimated number of compactions remaining to perform.            |
-| cassandra.storage.load               | UpDownCounter | By        |                                       | Size of the on disk data size this node manages.                 |
-| cassandra.storage.hints.count        | Counter       | {hint}    |                                       | Number of hint messages written to this node since server start. |
-| cassandra.storage.hints.in_progress  | UpDownCounter | {hint}    |                                       | Number of hints attempting to be sent currently.                 |
+| Metric Name                          | Type          | Unit      | Attributes                                                          | Description                                                      |
+| ------------------------------------ | ------------- | --------- | ------------------------------------------------------------------- | ---------------------------------------------------------------- |
+| cassandra.client.request.count       | Counter       | {request} | cassandra.operation                                                 | Number of requests by operation.                                 |
+| cassandra.client.request.error       | Counter       | {error}   | cassandra.operation, cassandra.status                               | Number of request errors by operation.                           |
+| cassandra.client.request.latency.p50 | Gauge         | s         | cassandra.operation                                                 | Request latency 50th percentile by operation.                    |
+| cassandra.client.request.latency.p99 | Gauge         | s         | cassandra.operation                                                 | Request latency 99th percentile by operation.                    |
+| cassandra.client.request.latency.max | Gauge         | s         | cassandra.operation                                                 | Maximum request latency by operation.                            |
+| cassandra.compaction.progress.bytes  | Gauge         | By        | cassandra.compaction.task_type, cassandra.keyspace, cassandra.table | Bytes completed for in-flight compactions.                       |
+| cassandra.compaction.progress.total  | Gauge         | By        | cassandra.compaction.task_type, cassandra.keyspace, cassandra.table | Total bytes for in-flight compactions.                           |
+| cassandra.compaction.tasks.completed | Counter       | {task}    |                                                                     | Number of completed compactions since server start.              |
+| cassandra.compaction.tasks.pending   | Gauge         | {task}    |                                                                     | Estimated number of compactions remaining to perform.            |
+| cassandra.storage.load               | UpDownCounter | By        |                                                                     | Size of the on disk data size this node manages.                 |
+| cassandra.storage.hints.count        | Counter       | {hint}    |                                                                     | Number of hint messages written to this node since server start. |
+| cassandra.storage.hints.in_progress  | UpDownCounter | {hint}    |                                                                     | Number of hints attempting to be sent currently.                 |

instrumentation/jmx-metrics/library/src/main/java/io/opentelemetry/instrumentation/jmx/internal/handler/CassandraCompactionProgressHandler.java

@@ -0,0 +1,174 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.jmx.internal.handler;
+
+import static java.util.logging.Level.WARNING;
+
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.api.metrics.Meter;
+import io.opentelemetry.api.metrics.ObservableLongMeasurement;
+import io.opentelemetry.instrumentation.jmx.internal.ExperimentalJmxMetricHandler;
+import java.math.BigInteger;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+import java.util.logging.Logger;
+import javax.annotation.Nullable;
+import javax.management.MBeanServerConnection;
+import javax.management.ObjectName;
+
+/**
+ * JMX metric handler that reports per-compaction byte progress from Cassandra's CompactionManager.
+ *
+ * <p>Queries {@code org.apache.cassandra.db:type=CompactionManager}, reads the {@code Compactions}
+ * attribute (a list of maps representing in-flight compactions), groups entries by {@code taskType
+ * + keyspace + columnfamily}, and emits two gauges per group:
+ *
+ * <ul>
+ *   <li>{@code cassandra.compaction.progress.bytes} — bytes completed so far
+ *   <li>{@code cassandra.compaction.progress.total} — total bytes for the compaction
+ * </ul>
+ *
+ * <p>Both metrics carry {@code cassandra.compaction.task_type}, {@code cassandra.keyspace}, and
+ * {@code cassandra.table} attributes. Entries missing any of these dimension fields are skipped.
+ * Byte values are string-encoded in the MBean and parsed via {@link BigInteger}.
+ *
+ * <p>Note: these are distinct from {@code cassandra.compaction.tasks.pending/completed}, which are
+ * simple scalar task counts from {@code org.apache.cassandra.metrics:type=Compaction}.
+ *
+ * <p>This class is internal and is hence not for public use. Its APIs are unstable and can change
+ * at any time.
+ */
+public class CassandraCompactionProgressHandler implements ExperimentalJmxMetricHandler {
+
+  static final String HANDLER_NAME = "cassandra-compaction-progress";
+  static final String METRIC_CURRENT = "cassandra.compaction.progress.bytes";
+  static final String METRIC_TOTAL = "cassandra.compaction.progress.total";
+
+  private static final String ATTR_TASK_TYPE = "cassandra.compaction.task_type";
+  private static final String ATTR_KEYSPACE = "cassandra.keyspace";
+  private static final String ATTR_TABLE = "cassandra.table";
+
+  private static final Logger logger =
+      Logger.getLogger(CassandraCompactionProgressHandler.class.getName());
+
+  @Override
+  public String getName() {
+    return HANDLER_NAME;
+  }
+
+  @Override
+  public AutoCloseable create(Meter meter, Supplier<Detector> detectorSupplier) {
+    ObservableLongMeasurement currentGauge =
+        meter
+            .gaugeBuilder(METRIC_CURRENT)
+            .setDescription("Bytes completed for in-flight compactions")
+            .setUnit("By")
+            .ofLongs()
+            .buildObserver();
+
+    ObservableLongMeasurement totalGauge =
+        meter
+            .gaugeBuilder(METRIC_TOTAL)
+            .setDescription("Total bytes for in-flight compactions")
+            .setUnit("By")
+            .ofLongs()
+            .buildObserver();
+
+    return meter.batchCallback(
+        () -> {
+          for (Map.Entry<Attributes, long[]> entry : queryGroups(detectorSupplier).entrySet()) {
+            currentGauge.record(entry.getValue()[0], entry.getKey());
+            totalGauge.record(entry.getValue()[1], entry.getKey());
+          }
+        },
+        currentGauge,
+        totalGauge);
+  }
+
+  private static Map<Attributes, long[]> queryGroups(Supplier<Detector> detectorSupplier) {
+    Map<Attributes, long[]> groups = new HashMap<>();
+    Detector detector = detectorSupplier.get();
+    if (detector == null) {
+      return groups;
+    }
+    MBeanServerConnection connection = detector.getConnection();
+    for (ObjectName objectName : detector.getObjectNames()) {
+      queryCompactions(connection, objectName)
+          .forEach(
+              (attrs, values) ->
+                  groups.merge(attrs, values, (a, b) -> new long[] {a[0] + b[0], a[1] + b[1]}));
+    }
+    return groups;
+  }
+
+  static Map<Attributes, long[]> queryCompactions(
+      MBeanServerConnection connection, ObjectName objectName) {
+    Map<Attributes, long[]> groups = new HashMap<>();
+    try {
+      // CompactionManager#getCompactions() returns List<Map<String,String>> via Standard MBean
+      @SuppressWarnings("unchecked")
+      List<Map<String, String>> compactions =
+          (List<Map<String, String>>) connection.getAttribute(objectName, "Compactions");
+
+      for (Map<String, String> entry : compactions) {
+        String taskType = entry.get("taskType");
+        String keyspace = entry.get("keyspace");
+        String columnFamily = entry.get("columnfamily");
+        String unit = entry.get("unit");
+
+        if (taskType == null || keyspace == null || columnFamily == null || !isByteUnit(unit)) {
+          continue;
+        }
+
+        long completed;
+        long total;
+        try {
+          completed = parseLong(entry.get("completed"));
+          total = parseLong(entry.get("total"));
+        } catch (ArithmeticException e) {
+          logger.log(
+              WARNING,
+              "cassandra.compaction.progress: byte value overflows long range, skipping entry",
+              e);
+          continue;
+        }
+
+        Attributes attrs = buildAttributes(taskType, keyspace, columnFamily);
+        groups.merge(
+            attrs, new long[] {completed, total}, (a, b) -> new long[] {a[0] + b[0], a[1] + b[1]});
+      }
+    } catch (Exception e) {
+      logger.log(WARNING, "cassandra.compaction.progress: failed to query CompactionManager", e);
+    }
+    return groups;
+  }
+
+  private static boolean isByteUnit(@Nullable String unit) {
+    return unit != null && unit.equalsIgnoreCase("bytes");
+  }
+
+  private static long parseLong(@Nullable Object value) {
+    if (value == null) {
+      return 0;
+    }
+    try {
+      return new BigInteger(value.toString()).longValueExact();
+    } catch (NumberFormatException e) {
+      return 0;
+    }
+  }
+
+  private static Attributes buildAttributes(String taskType, String keyspace, String columnFamily) {
+    AttributesBuilder builder = Attributes.builder();
+    builder.put(ATTR_TASK_TYPE, taskType);
+    builder.put(ATTR_KEYSPACE, keyspace);
+    builder.put(ATTR_TABLE, columnFamily);
+    return builder.build();
+  }
+}

instrumentation/jmx-metrics/library/src/main/resources/META-INF/services/io.opentelemetry.instrumentation.jmx.internal.ExperimentalJmxMetricHandler

@@ -0,0 +1 @@
+io.opentelemetry.instrumentation.jmx.internal.handler.CassandraCompactionProgressHandler

instrumentation/jmx-metrics/library/src/main/resources/jmx/rules/experimental-cassandra.yaml

@@ -120,3 +120,9 @@ rules:
         type: *errortype
         unit: *errorunit
         desc: *errordesc
+
+  # Compaction byte progress - uses a code-based handler because the Compactions attribute
+  # returns a list of maps requiring iteration, grouping by composite key, and string-encoded
+  # BigInteger parsing, none of which can be expressed in declarative YAML.
+  - bean: org.apache.cassandra.db:type=CompactionManager
+    handler: cassandra-compaction-progress

instrumentation/jmx-metrics/library/src/test/java/io/opentelemetry/instrumentation/jmx/internal/handler/CassandraCompactionProgressHandlerTest.java

@@ -0,0 +1,169 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.jmx.internal.handler;
+
+import static java.util.Arrays.asList;
+import static java.util.Collections.singletonList;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import io.opentelemetry.api.common.Attributes;
+import java.util.HashMap;
+import java.util.Map;
+import javax.management.MBeanServerConnection;
+import javax.management.ObjectName;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class CassandraCompactionProgressHandlerTest {
+
+  private static final String ATTR_TASK_TYPE = "cassandra.compaction.task_type";
+  private static final String ATTR_KEYSPACE = "cassandra.keyspace";
+  private static final String ATTR_TABLE = "cassandra.table";
+
+  private MBeanServerConnection connection;
+  private ObjectName objectName;
+
+  @BeforeEach
+  void setUp() throws Exception {
+    connection = mock(MBeanServerConnection.class);
+    objectName = new ObjectName("org.apache.cassandra.db:type=CompactionManager");
+  }
+
+  @Test
+  void groupsByCompositeKey() throws Exception {
+    when(connection.getAttribute(objectName, "Compactions"))
+        .thenReturn(
+            asList(
+                compactionEntry("COMPACTION", "ks1", "cf1", "100", "200"),
+                compactionEntry("COMPACTION", "ks1", "cf1", "50", "150"),
+                compactionEntry("COMPACTION", "ks2", "cf2", "10", "100")));
+
+    Map<Attributes, long[]> groups =
+        CassandraCompactionProgressHandler.queryCompactions(connection, objectName);
+
+    assertThat(groups).hasSize(2);
+    assertThat(groups.get(attrs("COMPACTION", "ks1", "cf1"))).containsExactly(150L, 350L);
+    assertThat(groups.get(attrs("COMPACTION", "ks2", "cf2"))).containsExactly(10L, 100L);
+  }
+
+  @Test
+  void skipsEntriesMissingDimensionFields() throws Exception {
+    when(connection.getAttribute(objectName, "Compactions"))
+        .thenReturn(
+            asList(
+                compactionEntry("COMPACTION", "ks1", "cf1", "10", "100"),
+                compactionEntry("COMPACTION", null, "cf1", "5", "50"),
+                compactionEntry("COMPACTION", "ks1", null, "5", "50")));
+
+    Map<Attributes, long[]> groups =
+        CassandraCompactionProgressHandler.queryCompactions(connection, objectName);
+
+    assertThat(groups).hasSize(1);
+    assertThat(groups.get(attrs("COMPACTION", "ks1", "cf1"))).containsExactly(10L, 100L);
+  }
+
+  @Test
+  void skipsEntriesWithNonByteUnits() throws Exception {
+    when(connection.getAttribute(objectName, "Compactions"))
+        .thenReturn(
+            asList(
+                compactionEntry("COMPACTION", "ks1", "cf1", "10", "100", "bytes"),
+                compactionEntry("VALIDATION", "ks1", "cf1", "5", "50", "keys"),
+                compactionEntry("ANTICOMPACTION", "ks1", "cf1", "5", "50", "ranges"),
+                compactionEntry("COMPACTION", "ks2", "cf2", "5", "50", null)));
+
+    Map<Attributes, long[]> groups =
+        CassandraCompactionProgressHandler.queryCompactions(connection, objectName);
+
+    assertThat(groups).hasSize(1);
+    assertThat(groups.get(attrs("COMPACTION", "ks1", "cf1"))).containsExactly(10L, 100L);
+  }
+
+  @Test
+  void skipsEntriesWithValuesExceedingLongRange() throws Exception {
+    // values larger than Long.MAX_VALUE cannot be safely cast to long — entry must be skipped
+    String big = "99999999999999999999";
+    when(connection.getAttribute(objectName, "Compactions"))
+        .thenReturn(singletonList(compactionEntry("COMPACTION", "ks", "cf", big, big)));
+
+    Map<Attributes, long[]> groups =
+        CassandraCompactionProgressHandler.queryCompactions(connection, objectName);
+
+    assertThat(groups).isEmpty();
+  }
+
+  @Test
+  void returnsEmptyMapOnException() throws Exception {
+    when(connection.getAttribute(objectName, "Compactions"))
+        .thenThrow(new RuntimeException("connection lost"));
+
+    Map<Attributes, long[]> groups =
+        CassandraCompactionProgressHandler.queryCompactions(connection, objectName);
+
+    assertThat(groups).isEmpty();
+  }
+
+  @Test
+  void handlerNameIsStable() {
+    assertThat(new CassandraCompactionProgressHandler().getName())
+        .isEqualTo(CassandraCompactionProgressHandler.HANDLER_NAME);
+  }
+
+  @Test
+  void mergesGroupsAcrossMultipleObjectNames() throws Exception {
+    ObjectName objectName2 = new ObjectName("org.apache.cassandra.db:type=CompactionManager,id=2");
+    when(connection.getAttribute(objectName, "Compactions"))
+        .thenReturn(singletonList(compactionEntry("COMPACTION", "ks1", "cf1", "100", "200")));
+    when(connection.getAttribute(objectName2, "Compactions"))
+        .thenReturn(singletonList(compactionEntry("COMPACTION", "ks1", "cf1", "50", "150")));
+
+    Map<Attributes, long[]> first =
+        CassandraCompactionProgressHandler.queryCompactions(connection, objectName);
+    Map<Attributes, long[]> second =
+        CassandraCompactionProgressHandler.queryCompactions(connection, objectName2);
+
+    // Simulate what queryGroups does: merge across ObjectNames using the same merge function
+    Map<Attributes, long[]> merged = new HashMap<>(first);
+    second.forEach(
+        (attrs, values) ->
+            merged.merge(attrs, values, (a, b) -> new long[] {a[0] + b[0], a[1] + b[1]}));
+
+    assertThat(merged).hasSize(1);
+    assertThat(merged.get(attrs("COMPACTION", "ks1", "cf1"))).containsExactly(150L, 350L);
+  }
+
+  private static Map<String, String> compactionEntry(
+      String taskType, String keyspace, String columnfamily, String completed, String total) {
+    return compactionEntry(taskType, keyspace, columnfamily, completed, total, "bytes");
+  }
+
+  private static Map<String, String> compactionEntry(
+      String taskType,
+      String keyspace,
+      String columnfamily,
+      String completed,
+      String total,
+      String unit) {
+    Map<String, String> entry = new HashMap<>();
+    entry.put("taskType", taskType);
+    entry.put("keyspace", keyspace);
+    entry.put("columnfamily", columnfamily);
+    entry.put("completed", completed);
+    entry.put("total", total);
+    entry.put("unit", unit);
+    return entry;
+  }
+
+  private static Attributes attrs(String taskType, String keyspace, String columnFamily) {
+    return Attributes.builder()
+        .put(ATTR_TASK_TYPE, taskType)
+        .put(ATTR_KEYSPACE, keyspace)
+        .put(ATTR_TABLE, columnFamily)
+        .build();
+  }
+}