frankgh commented on code in PR #249: URL: https://github.com/apache/cassandra-sidecar/pull/249#discussion_r2282289947
########## server/src/main/java/org/apache/cassandra/sidecar/handlers/CompactionStatsHandler.java: ########## @@ -0,0 +1,82 @@ +/* + * 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.cassandra.sidecar.handlers; + +import java.util.Collections; +import java.util.Set; + +import com.google.inject.Inject; +import io.vertx.core.http.HttpServerRequest; +import io.vertx.core.net.SocketAddress; +import io.vertx.ext.auth.authorization.Authorization; +import io.vertx.ext.web.RoutingContext; +import org.apache.cassandra.sidecar.acl.authorization.BasicPermissions; +import org.apache.cassandra.sidecar.acl.authorization.CassandraPermissions; +import org.apache.cassandra.sidecar.common.server.MetricsOperations; +import org.apache.cassandra.sidecar.concurrent.ExecutorPools; +import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher; +import org.jetbrains.annotations.NotNull; + +import static org.apache.cassandra.sidecar.acl.authorization.ResourceScopes.DATA_SCOPE; + +/** + * Handler for retrieving compaction statistics + */ +public class CompactionStatsHandler extends AbstractHandler<Void> implements AccessProtected +{ + /** + * Constructs a handler with the provided {@code metadataFetcher} + * + * @param metadataFetcher the metadata fetcher + * @param executorPools executor pools for blocking executions + */ + @Inject + protected CompactionStatsHandler(final InstanceMetadataFetcher metadataFetcher, final ExecutorPools executorPools) + { + super(metadataFetcher, executorPools, null); + } + + @Override + public Set<Authorization> requiredAuthorizations() + { + return Collections.singleton(BasicPermissions.STATS_CLUSTER_SCOPED.toAuthorization()); + } + + /** + * {@inheritDoc} + */ + @Override + public void handleInternal(RoutingContext context, + HttpServerRequest httpRequest, + @NotNull String host, + SocketAddress remoteAddress, + Void request) + { + MetricsOperations operations = metadataFetcher.delegate(host).metricsOperations(); + executorPools.service() + .executeBlocking(operations::compactionStats) + .onSuccess(context::json) + .onFailure(cause -> processFailure(cause, context, host, remoteAddress, request)); + } + + protected Void extractParamsOrThrow(RoutingContext context) Review Comment: include override annotation ```suggestion @Override protected Void extractParamsOrThrow(RoutingContext context) ``` ########## adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/CassandraMetricsOperations.java: ########## @@ -250,4 +295,186 @@ private List<ClientConnectionEntry> statsToEntries(Stream<ConnectedClientStats> stat.authenticationMode, stat.authenticationMetadata); } + + /** + * Represents the metrics related to compaction stats that are supported by the Sidecar + */ + public enum CompactionStatsMetrics + { + TOTAL_COMPACTIONS_COMPLETED("TotalCompactionsCompleted", MetricType.COUNTER), + BYTES_COMPACTED("BytesCompacted", MetricType.COUNTER), + COMPACTIONS_ABORTED("CompactionsAborted", MetricType.COUNTER), + COMPACTIONS_REDUCED("CompactionsReduced", MetricType.COUNTER), + SSTABLES_DROPPED_FROM_COMPACTION("SSTablesDroppedFromCompaction", MetricType.COUNTER), + PENDING_TASKS_BY_TABLE_NAME("PendingTasksByTableName", MetricType.GAUGE); + + private final String metricName; + private final MetricType type; + + CompactionStatsMetrics(String metricName, MetricType type) + { + this.metricName = metricName; + this.type = type; + } + + String metricName() + { + return metricName; + } + } + /** + * {@inheritDoc} + */ + @Override + public CompactionStatsResponse compactionStats() + { + // Get compaction manager and storage service proxies + CompactionManagerJmxOperations compactionManager = jmxClient.proxy(CompactionManagerJmxOperations.class, COMPACTION_MANAGER_OBJ_NAME); + StorageJmxOperations storageService = jmxClient.proxy(StorageJmxOperations.class, STORAGE_SERVICE_OBJ_NAME); + + // Get concurrent compactors from StorageService as per specification + long concurrentCompactors = storageService.getConcurrentCompactors(); + + // Get pending tasks grouped by keyspace and table + Map<String, Map<String, Integer>> pendingTasks = getPendingCompactionTasksByTable(); + long totalPendingTasks = pendingTasks.values().stream() + .mapToLong(tableMap -> tableMap.values().stream().mapToInt(Integer::intValue).sum()) + .sum(); + + // Get compaction metrics from JMX counters and meters + long completedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.TOTAL_COMPACTIONS_COMPLETED)); + long dataCompacted = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.BYTES_COMPACTED)); + long abortedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.COMPACTIONS_ABORTED)); + long reducedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.COMPACTIONS_REDUCED)); + long sstablesDroppedFromCompaction = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.SSTABLES_DROPPED_FROM_COMPACTION)); + + // Get completed compactions rate with proper time conversions + CompactionStatsResponse.CompletedCompactionsRate completedCompactionsRate = getCompletedCompactionsRate(); + + // Get active compactions with all required fields + List<ActiveCompactionEntry> activeCompactions = getActiveCompactions(compactionManager.getCompactions()); + long activeCompactionsCount = activeCompactions.size(); + + // Calculate remaining time in seconds based on throughput and remaining bytes + String activeCompactionsRemainingTime = calculateRemainingTimeSeconds(activeCompactions, storageService); + + return new CompactionStatsResponse(concurrentCompactors, pendingTasks, totalPendingTasks, + completedCompactions, dataCompacted, abortedCompactions, + reducedCompactions, sstablesDroppedFromCompaction, + completedCompactionsRate, + activeCompactions, activeCompactionsCount, activeCompactionsRemainingTime); + } + + private Map<String, Map<String, Integer>> getPendingCompactionTasksByTable() + { + // Get pending tasks by table name from Gauge metric + Object value = getCompactionMetric(CompactionStatsMetrics.PENDING_TASKS_BY_TABLE_NAME); + return parsePendingTasksMap(value); + } + + private Map<String, Map<String, Integer>> parsePendingTasksMap(final Object value) + { + Map<?, ?> rawMap = safeCast(value, Map.class, "pending tasks"); + Map<String, Map<String, Integer>> result = new HashMap<>(); + + for (Map.Entry<?, ?> entry : rawMap.entrySet()) + { + String keyspace = safeCast(entry.getKey(), String.class, "keyspace name"); + Map<?, ?> rawTableMap = safeCast(entry.getValue(), Map.class, "table data"); + Map<String, Integer> tableMap = new HashMap<>(); + + for (Map.Entry<?, ?> tableEntry : rawTableMap.entrySet()) + { + String tableName = safeCast(tableEntry.getKey(), String.class, "table name"); + Number taskCount = safeCast(tableEntry.getValue(), Number.class, "task count"); + tableMap.put(tableName, taskCount.intValue()); + } + + result.put(keyspace, tableMap); + } + + return result; + } + + private Object getCompactionMetric(final CompactionStatsMetrics metric) + { + String metricObjectType = String.format(METRICS_OBJ_TYPE_COMPACTION, metric.metricName()); + return queryMetric(metricObjectType, metric.type); + } + + private CompactionStatsResponse.CompletedCompactionsRate getCompletedCompactionsRate() + { + // Get rates from meter metric for TotalCompactionsCompleted + String metricObjectType = String.format(METRICS_OBJ_TYPE_COMPACTION, "TotalCompactionsCompleted"); + MeterMetricsJmxOperations metricsProxy = jmxClient.proxy(MeterMetricsJmxOperations.class, metricObjectType); + + // Convert per-second rates to the specification: + // meanRate: compactions per hour + // fifteenMinuteRate: compactions per minute for last 15 minutes + double meanRateValue = metricsProxy.getMeanRate() * 3600; // Convert per second to per hour + double fifteenMinuteRateValue = metricsProxy.getFifteenMinuteRate() * 60; // Convert per second to per minute + + String meanRate = String.format("%.2f/hour", meanRateValue); + String fifteenMinuteRate = String.format("%.2f/minute", fifteenMinuteRateValue); + + return new CompactionStatsResponse.CompletedCompactionsRate(meanRate, fifteenMinuteRate); + } + + + private List<ActiveCompactionEntry> getActiveCompactions(List<Map<String, String>> compactions) + { + return compactions.stream().map(compactionInfo -> { + // Extract fields according to specification + String id = compactionInfo.getOrDefault(COMPACTION_ID, DEFAULT_STRING_VALUE); + String keyspace = compactionInfo.getOrDefault(KEYSPACE, DEFAULT_STRING_VALUE); + String columnFamily = compactionInfo.getOrDefault(COLUMNFAMILY, DEFAULT_STRING_VALUE); + String taskType = compactionInfo.getOrDefault(TASK_TYPE, DEFAULT_STRING_VALUE); + + // Parse byte values + long completedBytes = safeParseLong(compactionInfo.getOrDefault(COMPLETED, DEFAULT_NUMBER_VALUE), "completed bytes"); + long totalBytes = safeParseLong(compactionInfo.getOrDefault(TOTAL, DEFAULT_NUMBER_VALUE), "total bytes"); + + // Calculate percentage completed + double percentCompleted = totalBytes == -1 ? 0.0 : (double) completedBytes / totalBytes * 100.0; Review Comment: if totalBytes == 0 then your division will throw ########## integration-tests/src/integrationTest/org/apache/cassandra/sidecar/routes/CompactionStatsIntegrationTest.java: ########## @@ -0,0 +1,372 @@ +/* + * 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.cassandra.sidecar.routes; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.jupiter.api.Test; + +import io.vertx.core.buffer.Buffer; +import io.vertx.core.http.HttpResponseExpectation; +import io.vertx.ext.web.client.HttpResponse; +import org.apache.cassandra.sidecar.common.response.CompactionStatsResponse; +import org.apache.cassandra.sidecar.common.response.data.ActiveCompactionEntry; +import org.apache.cassandra.sidecar.testing.QualifiedName; +import org.apache.cassandra.sidecar.testing.SharedClusterSidecarIntegrationTestBase; + +import static org.apache.cassandra.testing.TestUtils.DC1_RF1; +import static org.apache.cassandra.testing.TestUtils.TEST_KEYSPACE; +import static org.apache.cassandra.testing.TestUtils.TEST_TABLE_PREFIX; +import static org.apache.cassandra.testing.utils.AssertionUtils.getBlocking; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Integration tests for the Compaction Statistics API endpoint. + * + * <p>Tests the compaction statistics functionality by generating large-scale SSTable data across + * multiple tables, triggering concurrent compactions, and validating API responses in both active + * and completed compaction scenarios.</p> + * + * <h2>Testing Mechanism</h2> + * <ul> + * <li><strong>Data Setup:</strong> Creates 5 tables with 100 SSTables each (500 total, 1000 records per SSTable)</li> + * <li><strong>Concurrency:</strong> Triggers compactions simultaneously using separate threads</li> + * <li><strong>Active Detection:</strong> Polls API immediately with retry logic to catch in-progress compactions</li> + * <li><strong>Comprehensive Validation:</strong> Validates all response fields including statistics, rates, + * pending tasks, and detailed active compaction data when available</li> + * <li><strong>Dual Success:</strong> Test passes whether active compactions are captured or completed</li> + * </ul> + */ +class CompactionStatsIntegrationTest extends SharedClusterSidecarIntegrationTestBase Review Comment: we should merge this test class with `org.apache.cassandra.sidecar.routes.CassandraStatsIntegrationTest` ########## adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/CassandraMetricsOperations.java: ########## @@ -250,4 +295,186 @@ private List<ClientConnectionEntry> statsToEntries(Stream<ConnectedClientStats> stat.authenticationMode, stat.authenticationMetadata); } + + /** + * Represents the metrics related to compaction stats that are supported by the Sidecar + */ + public enum CompactionStatsMetrics + { + TOTAL_COMPACTIONS_COMPLETED("TotalCompactionsCompleted", MetricType.COUNTER), + BYTES_COMPACTED("BytesCompacted", MetricType.COUNTER), + COMPACTIONS_ABORTED("CompactionsAborted", MetricType.COUNTER), + COMPACTIONS_REDUCED("CompactionsReduced", MetricType.COUNTER), + SSTABLES_DROPPED_FROM_COMPACTION("SSTablesDroppedFromCompaction", MetricType.COUNTER), + PENDING_TASKS_BY_TABLE_NAME("PendingTasksByTableName", MetricType.GAUGE); + + private final String metricName; + private final MetricType type; + + CompactionStatsMetrics(String metricName, MetricType type) + { + this.metricName = metricName; + this.type = type; + } + + String metricName() + { + return metricName; + } + } + /** + * {@inheritDoc} + */ + @Override + public CompactionStatsResponse compactionStats() + { + // Get compaction manager and storage service proxies + CompactionManagerJmxOperations compactionManager = jmxClient.proxy(CompactionManagerJmxOperations.class, COMPACTION_MANAGER_OBJ_NAME); Review Comment: we should segregate the logic to get the compaction stats from this class. This class should be a thin layer that proxies the request into another class where you have the business logic to retrieve compaction stats ########## adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/CassandraMetricsOperations.java: ########## @@ -250,4 +295,186 @@ private List<ClientConnectionEntry> statsToEntries(Stream<ConnectedClientStats> stat.authenticationMode, stat.authenticationMetadata); } + + /** + * Represents the metrics related to compaction stats that are supported by the Sidecar + */ + public enum CompactionStatsMetrics Review Comment: this feels misplaced, move it to its own class file ########## adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/CassandraMetricsOperations.java: ########## Review Comment: generally seeing a lot of empty lines with whitespace, it would be good to format them ########## adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/CassandraMetricsOperations.java: ########## @@ -250,4 +295,186 @@ private List<ClientConnectionEntry> statsToEntries(Stream<ConnectedClientStats> stat.authenticationMode, stat.authenticationMetadata); } + + /** + * Represents the metrics related to compaction stats that are supported by the Sidecar + */ + public enum CompactionStatsMetrics + { + TOTAL_COMPACTIONS_COMPLETED("TotalCompactionsCompleted", MetricType.COUNTER), + BYTES_COMPACTED("BytesCompacted", MetricType.COUNTER), + COMPACTIONS_ABORTED("CompactionsAborted", MetricType.COUNTER), + COMPACTIONS_REDUCED("CompactionsReduced", MetricType.COUNTER), + SSTABLES_DROPPED_FROM_COMPACTION("SSTablesDroppedFromCompaction", MetricType.COUNTER), + PENDING_TASKS_BY_TABLE_NAME("PendingTasksByTableName", MetricType.GAUGE); + + private final String metricName; + private final MetricType type; + + CompactionStatsMetrics(String metricName, MetricType type) + { + this.metricName = metricName; + this.type = type; + } + + String metricName() + { + return metricName; + } + } + /** + * {@inheritDoc} + */ + @Override + public CompactionStatsResponse compactionStats() + { + // Get compaction manager and storage service proxies + CompactionManagerJmxOperations compactionManager = jmxClient.proxy(CompactionManagerJmxOperations.class, COMPACTION_MANAGER_OBJ_NAME); + StorageJmxOperations storageService = jmxClient.proxy(StorageJmxOperations.class, STORAGE_SERVICE_OBJ_NAME); + + // Get concurrent compactors from StorageService as per specification + long concurrentCompactors = storageService.getConcurrentCompactors(); + + // Get pending tasks grouped by keyspace and table + Map<String, Map<String, Integer>> pendingTasks = getPendingCompactionTasksByTable(); + long totalPendingTasks = pendingTasks.values().stream() + .mapToLong(tableMap -> tableMap.values().stream().mapToInt(Integer::intValue).sum()) + .sum(); + + // Get compaction metrics from JMX counters and meters + long completedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.TOTAL_COMPACTIONS_COMPLETED)); + long dataCompacted = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.BYTES_COMPACTED)); + long abortedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.COMPACTIONS_ABORTED)); + long reducedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.COMPACTIONS_REDUCED)); + long sstablesDroppedFromCompaction = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.SSTABLES_DROPPED_FROM_COMPACTION)); + + // Get completed compactions rate with proper time conversions + CompactionStatsResponse.CompletedCompactionsRate completedCompactionsRate = getCompletedCompactionsRate(); + + // Get active compactions with all required fields + List<ActiveCompactionEntry> activeCompactions = getActiveCompactions(compactionManager.getCompactions()); + long activeCompactionsCount = activeCompactions.size(); + + // Calculate remaining time in seconds based on throughput and remaining bytes + String activeCompactionsRemainingTime = calculateRemainingTimeSeconds(activeCompactions, storageService); + + return new CompactionStatsResponse(concurrentCompactors, pendingTasks, totalPendingTasks, + completedCompactions, dataCompacted, abortedCompactions, + reducedCompactions, sstablesDroppedFromCompaction, + completedCompactionsRate, + activeCompactions, activeCompactionsCount, activeCompactionsRemainingTime); + } + + private Map<String, Map<String, Integer>> getPendingCompactionTasksByTable() + { + // Get pending tasks by table name from Gauge metric + Object value = getCompactionMetric(CompactionStatsMetrics.PENDING_TASKS_BY_TABLE_NAME); + return parsePendingTasksMap(value); + } + + private Map<String, Map<String, Integer>> parsePendingTasksMap(final Object value) Review Comment: NIT ```suggestion private Map<String, Map<String, Integer>> parsePendingTasksMap(Object value) ``` ########## client-common/src/main/java/org/apache/cassandra/sidecar/common/response/data/ActiveCompactionEntry.java: ########## @@ -0,0 +1,131 @@ +/* + * 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.cassandra.sidecar.common.response.data; + +import java.util.List; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** + * Represents an active compaction entry + */ +@JsonInclude(JsonInclude.Include.NON_NULL) +public class ActiveCompactionEntry +{ + private final String id; + private final String keyspace; + private final String columnFamily; Review Comment: I think we should use table here instead. columFamily is the deprecated name, but it's probably kept here for backwards compatibility reasons. ########## adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/jmx/CompactionManagerJmxOperations.java: ########## @@ -0,0 +1,48 @@ +/* + * 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.cassandra.sidecar.adapters.base.jmx; + +import java.util.List; +import java.util.Map; + +/** + * An interface that pulls methods from the Cassandra CompactionManager JMX proxy + */ +public interface CompactionManagerJmxOperations +{ + String COMPACTION_MANAGER_OBJ_NAME = "org.apache.cassandra.db:type=CompactionManager"; + + /** + * Returns the number of concurrent compactors configured for the node. + * @return number of concurrent compactors + */ + int getCoreCompactorThreads(); Review Comment: unused, remove it? ########## adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/CassandraMetricsOperations.java: ########## @@ -250,4 +295,186 @@ private List<ClientConnectionEntry> statsToEntries(Stream<ConnectedClientStats> stat.authenticationMode, stat.authenticationMetadata); } + + /** + * Represents the metrics related to compaction stats that are supported by the Sidecar + */ + public enum CompactionStatsMetrics + { + TOTAL_COMPACTIONS_COMPLETED("TotalCompactionsCompleted", MetricType.COUNTER), + BYTES_COMPACTED("BytesCompacted", MetricType.COUNTER), + COMPACTIONS_ABORTED("CompactionsAborted", MetricType.COUNTER), + COMPACTIONS_REDUCED("CompactionsReduced", MetricType.COUNTER), + SSTABLES_DROPPED_FROM_COMPACTION("SSTablesDroppedFromCompaction", MetricType.COUNTER), + PENDING_TASKS_BY_TABLE_NAME("PendingTasksByTableName", MetricType.GAUGE); + + private final String metricName; + private final MetricType type; + + CompactionStatsMetrics(String metricName, MetricType type) + { + this.metricName = metricName; + this.type = type; + } + + String metricName() + { + return metricName; + } + } + /** + * {@inheritDoc} + */ + @Override + public CompactionStatsResponse compactionStats() + { + // Get compaction manager and storage service proxies + CompactionManagerJmxOperations compactionManager = jmxClient.proxy(CompactionManagerJmxOperations.class, COMPACTION_MANAGER_OBJ_NAME); + StorageJmxOperations storageService = jmxClient.proxy(StorageJmxOperations.class, STORAGE_SERVICE_OBJ_NAME); + + // Get concurrent compactors from StorageService as per specification + long concurrentCompactors = storageService.getConcurrentCompactors(); + + // Get pending tasks grouped by keyspace and table + Map<String, Map<String, Integer>> pendingTasks = getPendingCompactionTasksByTable(); + long totalPendingTasks = pendingTasks.values().stream() + .mapToLong(tableMap -> tableMap.values().stream().mapToInt(Integer::intValue).sum()) + .sum(); + + // Get compaction metrics from JMX counters and meters + long completedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.TOTAL_COMPACTIONS_COMPLETED)); + long dataCompacted = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.BYTES_COMPACTED)); Review Comment: let's preserve the units in the variable name ########## adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/CassandraMetricsOperations.java: ########## @@ -250,4 +295,186 @@ private List<ClientConnectionEntry> statsToEntries(Stream<ConnectedClientStats> stat.authenticationMode, stat.authenticationMetadata); } + + /** + * Represents the metrics related to compaction stats that are supported by the Sidecar + */ + public enum CompactionStatsMetrics + { + TOTAL_COMPACTIONS_COMPLETED("TotalCompactionsCompleted", MetricType.COUNTER), + BYTES_COMPACTED("BytesCompacted", MetricType.COUNTER), + COMPACTIONS_ABORTED("CompactionsAborted", MetricType.COUNTER), + COMPACTIONS_REDUCED("CompactionsReduced", MetricType.COUNTER), + SSTABLES_DROPPED_FROM_COMPACTION("SSTablesDroppedFromCompaction", MetricType.COUNTER), + PENDING_TASKS_BY_TABLE_NAME("PendingTasksByTableName", MetricType.GAUGE); + + private final String metricName; + private final MetricType type; + + CompactionStatsMetrics(String metricName, MetricType type) + { + this.metricName = metricName; + this.type = type; + } + + String metricName() + { + return metricName; + } + } + /** + * {@inheritDoc} + */ + @Override + public CompactionStatsResponse compactionStats() + { + // Get compaction manager and storage service proxies + CompactionManagerJmxOperations compactionManager = jmxClient.proxy(CompactionManagerJmxOperations.class, COMPACTION_MANAGER_OBJ_NAME); + StorageJmxOperations storageService = jmxClient.proxy(StorageJmxOperations.class, STORAGE_SERVICE_OBJ_NAME); + + // Get concurrent compactors from StorageService as per specification + long concurrentCompactors = storageService.getConcurrentCompactors(); + + // Get pending tasks grouped by keyspace and table + Map<String, Map<String, Integer>> pendingTasks = getPendingCompactionTasksByTable(); + long totalPendingTasks = pendingTasks.values().stream() + .mapToLong(tableMap -> tableMap.values().stream().mapToInt(Integer::intValue).sum()) + .sum(); + + // Get compaction metrics from JMX counters and meters + long completedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.TOTAL_COMPACTIONS_COMPLETED)); + long dataCompacted = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.BYTES_COMPACTED)); + long abortedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.COMPACTIONS_ABORTED)); + long reducedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.COMPACTIONS_REDUCED)); + long sstablesDroppedFromCompaction = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.SSTABLES_DROPPED_FROM_COMPACTION)); + + // Get completed compactions rate with proper time conversions + CompactionStatsResponse.CompletedCompactionsRate completedCompactionsRate = getCompletedCompactionsRate(); + + // Get active compactions with all required fields + List<ActiveCompactionEntry> activeCompactions = getActiveCompactions(compactionManager.getCompactions()); + long activeCompactionsCount = activeCompactions.size(); + + // Calculate remaining time in seconds based on throughput and remaining bytes + String activeCompactionsRemainingTime = calculateRemainingTimeSeconds(activeCompactions, storageService); + + return new CompactionStatsResponse(concurrentCompactors, pendingTasks, totalPendingTasks, + completedCompactions, dataCompacted, abortedCompactions, + reducedCompactions, sstablesDroppedFromCompaction, + completedCompactionsRate, + activeCompactions, activeCompactionsCount, activeCompactionsRemainingTime); + } + + private Map<String, Map<String, Integer>> getPendingCompactionTasksByTable() + { + // Get pending tasks by table name from Gauge metric + Object value = getCompactionMetric(CompactionStatsMetrics.PENDING_TASKS_BY_TABLE_NAME); + return parsePendingTasksMap(value); + } + + private Map<String, Map<String, Integer>> parsePendingTasksMap(final Object value) + { + Map<?, ?> rawMap = safeCast(value, Map.class, "pending tasks"); + Map<String, Map<String, Integer>> result = new HashMap<>(); + + for (Map.Entry<?, ?> entry : rawMap.entrySet()) + { + String keyspace = safeCast(entry.getKey(), String.class, "keyspace name"); + Map<?, ?> rawTableMap = safeCast(entry.getValue(), Map.class, "table data"); + Map<String, Integer> tableMap = new HashMap<>(); + + for (Map.Entry<?, ?> tableEntry : rawTableMap.entrySet()) + { + String tableName = safeCast(tableEntry.getKey(), String.class, "table name"); + Number taskCount = safeCast(tableEntry.getValue(), Number.class, "task count"); + tableMap.put(tableName, taskCount.intValue()); + } + + result.put(keyspace, tableMap); + } + + return result; + } + + private Object getCompactionMetric(final CompactionStatsMetrics metric) + { + String metricObjectType = String.format(METRICS_OBJ_TYPE_COMPACTION, metric.metricName()); + return queryMetric(metricObjectType, metric.type); + } + + private CompactionStatsResponse.CompletedCompactionsRate getCompletedCompactionsRate() + { + // Get rates from meter metric for TotalCompactionsCompleted + String metricObjectType = String.format(METRICS_OBJ_TYPE_COMPACTION, "TotalCompactionsCompleted"); + MeterMetricsJmxOperations metricsProxy = jmxClient.proxy(MeterMetricsJmxOperations.class, metricObjectType); + + // Convert per-second rates to the specification: + // meanRate: compactions per hour + // fifteenMinuteRate: compactions per minute for last 15 minutes + double meanRateValue = metricsProxy.getMeanRate() * 3600; // Convert per second to per hour + double fifteenMinuteRateValue = metricsProxy.getFifteenMinuteRate() * 60; // Convert per second to per minute + + String meanRate = String.format("%.2f/hour", meanRateValue); + String fifteenMinuteRate = String.format("%.2f/minute", fifteenMinuteRateValue); + + return new CompactionStatsResponse.CompletedCompactionsRate(meanRate, fifteenMinuteRate); + } + + + private List<ActiveCompactionEntry> getActiveCompactions(List<Map<String, String>> compactions) + { + return compactions.stream().map(compactionInfo -> { + // Extract fields according to specification + String id = compactionInfo.getOrDefault(COMPACTION_ID, DEFAULT_STRING_VALUE); + String keyspace = compactionInfo.getOrDefault(KEYSPACE, DEFAULT_STRING_VALUE); + String columnFamily = compactionInfo.getOrDefault(COLUMNFAMILY, DEFAULT_STRING_VALUE); + String taskType = compactionInfo.getOrDefault(TASK_TYPE, DEFAULT_STRING_VALUE); + + // Parse byte values + long completedBytes = safeParseLong(compactionInfo.getOrDefault(COMPLETED, DEFAULT_NUMBER_VALUE), "completed bytes"); + long totalBytes = safeParseLong(compactionInfo.getOrDefault(TOTAL, DEFAULT_NUMBER_VALUE), "total bytes"); + + // Calculate percentage completed + double percentCompleted = totalBytes == -1 ? 0.0 : (double) completedBytes / totalBytes * 100.0; + + // Parse SSTables list + String ssTablesStr = compactionInfo.getOrDefault(SSTABLES, DEFAULT_STRING_VALUE); + List<String> ssTables = ssTablesStr.isEmpty() ? List.of() : List.of(ssTablesStr.split(",")); + + String targetDirectory = compactionInfo.getOrDefault(TARGET_DIRECTORY, DEFAULT_STRING_VALUE); + + return new ActiveCompactionEntry(id, keyspace, columnFamily, taskType, + completedBytes, totalBytes, percentCompleted, + ssTables, targetDirectory); + }).collect(Collectors.toList()); + } + + private String calculateRemainingTimeSeconds(final List<ActiveCompactionEntry> activeCompactions, + final StorageJmxOperations storageService) Review Comment: NIT remove final keyword ```suggestion private String calculateRemainingTimeSeconds(List<ActiveCompactionEntry> activeCompactions, StorageJmxOperations storageService) ``` ########## adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/CassandraMetricsOperations.java: ########## @@ -250,4 +295,186 @@ private List<ClientConnectionEntry> statsToEntries(Stream<ConnectedClientStats> stat.authenticationMode, stat.authenticationMetadata); } + + /** + * Represents the metrics related to compaction stats that are supported by the Sidecar + */ + public enum CompactionStatsMetrics + { + TOTAL_COMPACTIONS_COMPLETED("TotalCompactionsCompleted", MetricType.COUNTER), + BYTES_COMPACTED("BytesCompacted", MetricType.COUNTER), + COMPACTIONS_ABORTED("CompactionsAborted", MetricType.COUNTER), + COMPACTIONS_REDUCED("CompactionsReduced", MetricType.COUNTER), + SSTABLES_DROPPED_FROM_COMPACTION("SSTablesDroppedFromCompaction", MetricType.COUNTER), + PENDING_TASKS_BY_TABLE_NAME("PendingTasksByTableName", MetricType.GAUGE); + + private final String metricName; + private final MetricType type; + + CompactionStatsMetrics(String metricName, MetricType type) + { + this.metricName = metricName; + this.type = type; + } + + String metricName() + { + return metricName; + } + } + /** + * {@inheritDoc} + */ + @Override + public CompactionStatsResponse compactionStats() + { + // Get compaction manager and storage service proxies + CompactionManagerJmxOperations compactionManager = jmxClient.proxy(CompactionManagerJmxOperations.class, COMPACTION_MANAGER_OBJ_NAME); + StorageJmxOperations storageService = jmxClient.proxy(StorageJmxOperations.class, STORAGE_SERVICE_OBJ_NAME); + + // Get concurrent compactors from StorageService as per specification + long concurrentCompactors = storageService.getConcurrentCompactors(); + + // Get pending tasks grouped by keyspace and table + Map<String, Map<String, Integer>> pendingTasks = getPendingCompactionTasksByTable(); + long totalPendingTasks = pendingTasks.values().stream() + .mapToLong(tableMap -> tableMap.values().stream().mapToInt(Integer::intValue).sum()) + .sum(); + + // Get compaction metrics from JMX counters and meters + long completedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.TOTAL_COMPACTIONS_COMPLETED)); + long dataCompacted = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.BYTES_COMPACTED)); + long abortedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.COMPACTIONS_ABORTED)); + long reducedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.COMPACTIONS_REDUCED)); + long sstablesDroppedFromCompaction = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.SSTABLES_DROPPED_FROM_COMPACTION)); + + // Get completed compactions rate with proper time conversions + CompactionStatsResponse.CompletedCompactionsRate completedCompactionsRate = getCompletedCompactionsRate(); + + // Get active compactions with all required fields + List<ActiveCompactionEntry> activeCompactions = getActiveCompactions(compactionManager.getCompactions()); + long activeCompactionsCount = activeCompactions.size(); + + // Calculate remaining time in seconds based on throughput and remaining bytes + String activeCompactionsRemainingTime = calculateRemainingTimeSeconds(activeCompactions, storageService); + + return new CompactionStatsResponse(concurrentCompactors, pendingTasks, totalPendingTasks, + completedCompactions, dataCompacted, abortedCompactions, + reducedCompactions, sstablesDroppedFromCompaction, + completedCompactionsRate, + activeCompactions, activeCompactionsCount, activeCompactionsRemainingTime); + } + + private Map<String, Map<String, Integer>> getPendingCompactionTasksByTable() + { + // Get pending tasks by table name from Gauge metric + Object value = getCompactionMetric(CompactionStatsMetrics.PENDING_TASKS_BY_TABLE_NAME); + return parsePendingTasksMap(value); + } + + private Map<String, Map<String, Integer>> parsePendingTasksMap(final Object value) + { + Map<?, ?> rawMap = safeCast(value, Map.class, "pending tasks"); + Map<String, Map<String, Integer>> result = new HashMap<>(); + + for (Map.Entry<?, ?> entry : rawMap.entrySet()) + { + String keyspace = safeCast(entry.getKey(), String.class, "keyspace name"); + Map<?, ?> rawTableMap = safeCast(entry.getValue(), Map.class, "table data"); + Map<String, Integer> tableMap = new HashMap<>(); + + for (Map.Entry<?, ?> tableEntry : rawTableMap.entrySet()) + { + String tableName = safeCast(tableEntry.getKey(), String.class, "table name"); + Number taskCount = safeCast(tableEntry.getValue(), Number.class, "task count"); + tableMap.put(tableName, taskCount.intValue()); + } + + result.put(keyspace, tableMap); + } + + return result; + } + + private Object getCompactionMetric(final CompactionStatsMetrics metric) + { + String metricObjectType = String.format(METRICS_OBJ_TYPE_COMPACTION, metric.metricName()); + return queryMetric(metricObjectType, metric.type); + } + + private CompactionStatsResponse.CompletedCompactionsRate getCompletedCompactionsRate() + { + // Get rates from meter metric for TotalCompactionsCompleted + String metricObjectType = String.format(METRICS_OBJ_TYPE_COMPACTION, "TotalCompactionsCompleted"); + MeterMetricsJmxOperations metricsProxy = jmxClient.proxy(MeterMetricsJmxOperations.class, metricObjectType); + + // Convert per-second rates to the specification: + // meanRate: compactions per hour + // fifteenMinuteRate: compactions per minute for last 15 minutes + double meanRateValue = metricsProxy.getMeanRate() * 3600; // Convert per second to per hour + double fifteenMinuteRateValue = metricsProxy.getFifteenMinuteRate() * 60; // Convert per second to per minute + + String meanRate = String.format("%.2f/hour", meanRateValue); + String fifteenMinuteRate = String.format("%.2f/minute", fifteenMinuteRateValue); + + return new CompactionStatsResponse.CompletedCompactionsRate(meanRate, fifteenMinuteRate); + } + + + private List<ActiveCompactionEntry> getActiveCompactions(List<Map<String, String>> compactions) + { + return compactions.stream().map(compactionInfo -> { + // Extract fields according to specification + String id = compactionInfo.getOrDefault(COMPACTION_ID, DEFAULT_STRING_VALUE); Review Comment: not sure why we are defaulting to empty string here? should it just be null instead? ########## adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/jmx/CounterMetricsJmxOperations.java: ########## @@ -21,16 +21,8 @@ /** * An interface that pulls methods from the Cassandra Metrics Proxy */ -public interface MetricsJmxOperations +public interface CounterMetricsJmxOperations Review Comment: not sure if I understand the motivation to split this interface into two, but it seems like keeping the original interface is sufficient? ########## adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/data/CompositeDataUtil.java: ########## @@ -50,4 +50,55 @@ public static <T> T extractValue(CompositeData data, String key) throw new RuntimeException("Value type mismatched of key: " + key, cce); } } + + /** + * Safely casts an object to the specified type with descriptive error handling. + * This method performs a runtime type check before casting to prevent ClassCastException + * and provides meaningful error messages when the cast fails. + * + * @param value the object to be cast + * @param expectedType the expected type to cast to + * @param contextDescription descriptive context for error messages (e.g., "keyspace name", "table data") + * @param <T> the target type + * @return the cast object of type T + * @throws IllegalStateException if the value is not an instance of the expected type, + * with a descriptive message indicating what was expected vs what was received + */ + public static <T> T safeCast(Object value, Class<T> expectedType, String contextDescription) Review Comment: we should have unit tests here. I was trying to understand what happens when you do `expectedType.isInstance(null)`. A unit test should give me an idea of what the behavior will be ########## adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/CassandraMetricsOperations.java: ########## @@ -250,4 +295,186 @@ private List<ClientConnectionEntry> statsToEntries(Stream<ConnectedClientStats> stat.authenticationMode, stat.authenticationMetadata); } + + /** + * Represents the metrics related to compaction stats that are supported by the Sidecar + */ + public enum CompactionStatsMetrics + { + TOTAL_COMPACTIONS_COMPLETED("TotalCompactionsCompleted", MetricType.COUNTER), + BYTES_COMPACTED("BytesCompacted", MetricType.COUNTER), + COMPACTIONS_ABORTED("CompactionsAborted", MetricType.COUNTER), + COMPACTIONS_REDUCED("CompactionsReduced", MetricType.COUNTER), + SSTABLES_DROPPED_FROM_COMPACTION("SSTablesDroppedFromCompaction", MetricType.COUNTER), + PENDING_TASKS_BY_TABLE_NAME("PendingTasksByTableName", MetricType.GAUGE); + + private final String metricName; + private final MetricType type; + + CompactionStatsMetrics(String metricName, MetricType type) + { + this.metricName = metricName; + this.type = type; + } + + String metricName() + { + return metricName; + } + } + /** + * {@inheritDoc} + */ + @Override + public CompactionStatsResponse compactionStats() + { + // Get compaction manager and storage service proxies + CompactionManagerJmxOperations compactionManager = jmxClient.proxy(CompactionManagerJmxOperations.class, COMPACTION_MANAGER_OBJ_NAME); + StorageJmxOperations storageService = jmxClient.proxy(StorageJmxOperations.class, STORAGE_SERVICE_OBJ_NAME); + + // Get concurrent compactors from StorageService as per specification + long concurrentCompactors = storageService.getConcurrentCompactors(); + + // Get pending tasks grouped by keyspace and table + Map<String, Map<String, Integer>> pendingTasks = getPendingCompactionTasksByTable(); + long totalPendingTasks = pendingTasks.values().stream() + .mapToLong(tableMap -> tableMap.values().stream().mapToInt(Integer::intValue).sum()) + .sum(); + + // Get compaction metrics from JMX counters and meters + long completedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.TOTAL_COMPACTIONS_COMPLETED)); + long dataCompacted = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.BYTES_COMPACTED)); + long abortedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.COMPACTIONS_ABORTED)); + long reducedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.COMPACTIONS_REDUCED)); + long sstablesDroppedFromCompaction = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.SSTABLES_DROPPED_FROM_COMPACTION)); + + // Get completed compactions rate with proper time conversions + CompactionStatsResponse.CompletedCompactionsRate completedCompactionsRate = getCompletedCompactionsRate(); + + // Get active compactions with all required fields + List<ActiveCompactionEntry> activeCompactions = getActiveCompactions(compactionManager.getCompactions()); + long activeCompactionsCount = activeCompactions.size(); + + // Calculate remaining time in seconds based on throughput and remaining bytes + String activeCompactionsRemainingTime = calculateRemainingTimeSeconds(activeCompactions, storageService); Review Comment: this should be a long value. And the formatting should happen at the client side. Your raw values should be preserved in the payload you send to the client ########## adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/jmx/CompactionManagerJmxOperations.java: ########## @@ -0,0 +1,48 @@ +/* + * 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.cassandra.sidecar.adapters.base.jmx; + +import java.util.List; +import java.util.Map; + +/** + * An interface that pulls methods from the Cassandra CompactionManager JMX proxy + */ +public interface CompactionManagerJmxOperations +{ + String COMPACTION_MANAGER_OBJ_NAME = "org.apache.cassandra.db:type=CompactionManager"; + + /** + * Returns the number of concurrent compactors configured for the node. + * @return number of concurrent compactors + */ + int getCoreCompactorThreads(); + + /** + * Returns active compactions as a list of compaction info maps + * @return list of compaction info maps + */ + List<Map<String, String>> getCompactions(); + + /** + * Returns compaction throughput in bytes per second Review Comment: it looks like the javadoc here is incorrect. Also, this method is not used. Do we need it? ########## adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/jmx/MeterMetricsJmxOperations.java: ########## @@ -0,0 +1,57 @@ +/* + * 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.cassandra.sidecar.adapters.base.jmx; + +/** + * An interface that pulls meter metric methods from Cassandra JMX proxy. + * Meter metrics track the rate of events occurring over time. + */ +public interface MeterMetricsJmxOperations +{ + + /** + * Returns the total number of events that have occurred. + * @return the total count of events + */ + long getCount(); + + /** + * Returns the mean rate of events per second over the entire lifetime of the meter. + * @return the mean rate in events per second + */ + double getMeanRate(); + + /** + * Returns the one-minute exponentially-weighted moving average rate. + * @return the one-minute rate in events per second + */ + double getOneMinuteRate(); + + /** + * Returns the five-minute exponentially-weighted moving average rate. + * @return the five-minute rate in events per second + */ + double getFiveMinuteRate(); Review Comment: unused ########## client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClient.java: ########## @@ -734,6 +735,20 @@ public CompletableFuture<ConnectedClientStatsResponse> connectedClientStats(Side .build()); } + /** + * Executes the compaction stats request using the default retry policy and provided {@code instance}. + * + * @param instance the instance where the request will be executed + * @return a completable future of the compaction stats + */ + public CompletableFuture<CompactionStatsResponse> compactionStats(SidecarInstance instance) Review Comment: can you add a test in `org.apache.cassandra.sidecar.client.SidecarClientTest` ########## adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/jmx/MeterMetricsJmxOperations.java: ########## @@ -0,0 +1,57 @@ +/* + * 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.cassandra.sidecar.adapters.base.jmx; + +/** + * An interface that pulls meter metric methods from Cassandra JMX proxy. + * Meter metrics track the rate of events occurring over time. + */ +public interface MeterMetricsJmxOperations +{ + + /** + * Returns the total number of events that have occurred. + * @return the total count of events + */ + long getCount(); + + /** + * Returns the mean rate of events per second over the entire lifetime of the meter. + * @return the mean rate in events per second + */ + double getMeanRate(); + + /** + * Returns the one-minute exponentially-weighted moving average rate. + * @return the one-minute rate in events per second + */ + double getOneMinuteRate(); Review Comment: unused ########## adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/CassandraMetricsOperations.java: ########## @@ -250,4 +278,187 @@ private List<ClientConnectionEntry> statsToEntries(Stream<ConnectedClientStats> stat.authenticationMode, stat.authenticationMetadata); } + + /** + * Represents the metrics related to compaction stats that are supported by the Sidecar + */ + public enum CompactionStatsMetrics + { + TOTAL_COMPACTIONS_COMPLETED("TotalCompactionsCompleted", MetricType.COUNTER), + BYTES_COMPACTED("BytesCompacted", MetricType.COUNTER), + COMPACTIONS_ABORTED("CompactionsAborted", MetricType.COUNTER), + COMPACTIONS_REDUCED("CompactionsReduced", MetricType.COUNTER), + SSTABLES_DROPPED_FROM_COMPACTION("SSTablesDroppedFromCompaction", MetricType.COUNTER), + PENDING_TASKS_BY_TABLE_NAME("PendingTasksByTableName", MetricType.GAUGE); + + private final String metricName; + private final MetricType type; + + CompactionStatsMetrics(String metricName, MetricType type) + { + this.metricName = metricName; + this.type = type; + } + + String metricName() + { + return metricName; + } + } + /** + * {@inheritDoc} + */ + @Override + public CompactionStatsResponse compactionStats() + { + // Get compaction manager and storage service proxies + CompactionManagerJmxOperations compactionManager = jmxClient.proxy(CompactionManagerJmxOperations.class, COMPACTION_MANAGER_OBJ_NAME); + StorageJmxOperations storageService = jmxClient.proxy(StorageJmxOperations.class, STORAGE_SERVICE_OBJ_NAME); + + // Get concurrent compactors from StorageService as per specification + long concurrentCompactors = storageService.getConcurrentCompactors(); + + // Get pending tasks grouped by keyspace and table + Map<String, Map<String, Integer>> pendingTasks = getPendingCompactionTasksByTable(); + long totalPendingTasks = pendingTasks.values().stream() + .mapToLong(tableMap -> tableMap.values().stream().mapToInt(Integer::intValue).sum()) + .sum(); + + // Get compaction metrics from JMX counters and meters + long completedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.TOTAL_COMPACTIONS_COMPLETED)); + long dataCompacted = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.BYTES_COMPACTED)); + long abortedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.COMPACTIONS_ABORTED)); + long reducedCompactions = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.COMPACTIONS_REDUCED)); + long sstablesDroppedFromCompaction = getValueAsLong(getCompactionMetric(CompactionStatsMetrics.SSTABLES_DROPPED_FROM_COMPACTION)); + + // Get completed compactions rate with proper time conversions + CompactionStatsResponse.CompletedCompactionsRate completedCompactionsRate = getCompletedCompactionsRate(); + + // Get active compactions with all required fields + List<ActiveCompactionEntry> activeCompactions = getActiveCompactions(compactionManager.getCompactions()); + long activeCompactionsCount = activeCompactions.size(); + + // Calculate remaining time in seconds based on throughput and remaining bytes + String activeCompactionsRemainingTime = calculateRemainingTimeSeconds(activeCompactions, storageService); + + return new CompactionStatsResponse(concurrentCompactors, pendingTasks, totalPendingTasks, + completedCompactions, dataCompacted, abortedCompactions, + reducedCompactions, sstablesDroppedFromCompaction, + completedCompactionsRate, + activeCompactions, activeCompactionsCount, activeCompactionsRemainingTime); Review Comment: I agree with Yifan here. We shouldn't leak the http layer to the adapter layer. Also, can you leverage the builder pattern here by leveraging `org.apache.cassandra.sidecar.common.DataObjectBuilder` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: pr-unsubscr...@cassandra.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: pr-unsubscr...@cassandra.apache.org For additional commands, e-mail: pr-h...@cassandra.apache.org
