diff --git a/CHANGES.txt b/CHANGES.txt index b32f2fce0b..f80e7ae8f0 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 5.1 + * Improved observability in AutoRepair to report both expected vs. actual repair bytes and expected vs. actual keyspaces (CASSANDRA-20581) * Execution of CreateTriggerStatement should not rely on external state (CASSANDRA-20287) * Support LIKE expressions in filtering queries (CASSANDRA-17198) * Make legacy index rebuilds safe on Gossip -> TCM upgrades (CASSANDRA-20887) diff --git a/doc/modules/cassandra/pages/managing/operating/metrics.adoc b/doc/modules/cassandra/pages/managing/operating/metrics.adoc index 2dc6d4fd99..ddc2ac851b 100644 --- a/doc/modules/cassandra/pages/managing/operating/metrics.adoc +++ b/doc/modules/cassandra/pages/managing/operating/metrics.adoc @@ -1085,40 +1085,58 @@ Reported name format: |=== |Name |Type |Description |RepairsInProgress |Gauge |Repair is in progress -on the node +on the node. |NodeRepairTimeInSec |Gauge |Time taken to repair -the node in seconds +the node in seconds. |ClusterRepairTimeInSec |Gauge |Time taken to repair -the entire Cassandra cluster in seconds +the entire Cassandra cluster in seconds. |LongestUnrepairedSec |Gauge |Time since the last repair -ran on the node in seconds +ran on the node in seconds. |RepairStartLagSec|Gauge |If a repair has not run within min_repair_interval, how long past this value since repairs last completed. Useful for determining if repairs are behind schedule. -|SucceededTokenRangesCount |Gauge |Number of token ranges successfully repaired on the node +|SucceededTokenRangesCount |Gauge |Number of token ranges successfully repaired on the node. -|FailedTokenRangesCount |Gauge |Number of token ranges failed to repair on the node +|FailedTokenRangesCount |Gauge |Number of token ranges failed to repair on the node. |SkippedTokenRangesCount |Gauge |Number of token ranges skipped -on the node +on the node. |SkippedTablesCount |Gauge |Number of tables skipped -on the node +on the node. |TotalMVTablesConsideredForRepair |Gauge |Number of materialized -views considered on the node +views considered on the node. |TotalDisabledRepairTables |Gauge |Number of tables on which -the automated repair has been disabled on the node +the automated repair has been disabled on the node. -|RepairTurnMyTurn |Counter |Represents the node's turn to repair +|TotalBytesToRepair |Gauge |Total bytes to be repaired across all keyspaces and tables involved in the current +repair schedule. + +|BytesAlreadyRepaired |Gauge |Cumulative number of bytes successfully repaired so far in the current +repair schedule. +NOTE: This calculation is the best effort for the FixedSplitTokenRangeSplitter. In practice, this metric +may not give you an accurate view in case of uneven data distribution. + + +|TotalKeyspaceRepairPlansToRepair |Gauge |Represents the total number of keyspace-level repair plans scheduled +for execution. If no table-level repair priorities are configured, this number typically matches the total number +of keyspaces under repair. However, if certain tables have repair priorities set, this number is usually higher than +the number of keyspaces, as multiple repair plans may be generated for different prioritized tables within the +same keyspace. + +|KeyspaceRepairPlansAlreadyRepaired |Gauge |Cumulative number of keyspace-level repair plans successfully +repaired so far in the current repair schedule. + +|RepairTurnMyTurn |Counter |Represents the node's turn to repair. |RepairTurnMyTurnDueToPriority |Counter |Represents the node's turn to repair -due to priority set in the automated repair +due to priority set in the automated repair. |RepairDelayedByReplica |Counter |Represents occurrences of a node's turn being delayed because a replica was currently taking its turn. Only relevant if diff --git a/src/java/org/apache/cassandra/metrics/AutoRepairMetrics.java b/src/java/org/apache/cassandra/metrics/AutoRepairMetrics.java index 3ef24a9eec..77d03c1c15 100644 --- a/src/java/org/apache/cassandra/metrics/AutoRepairMetrics.java +++ b/src/java/org/apache/cassandra/metrics/AutoRepairMetrics.java @@ -46,6 +46,11 @@ public class AutoRepairMetrics public final Gauge skippedTablesCount; public final Gauge totalMVTablesConsideredForRepair; public final Gauge totalDisabledRepairTables; + public final Gauge totalBytesToRepair; + public final Gauge bytesAlreadyRepaired; + public final Gauge totalKeyspaceRepairPlansToRepair; + public final Gauge keyspaceRepairPlansAlreadyRepaired; + public Counter repairTurnMyTurn; public Counter repairTurnMyTurnDueToPriority; public Counter repairTurnMyTurnForceRepair; @@ -155,6 +160,34 @@ public class AutoRepairMetrics return AutoRepair.instance.getRepairState(repairType).getTotalDisabledTablesRepairCount(); } }); + totalBytesToRepair = Metrics.register(factory.createMetricName("TotalBytesToRepair"), new Gauge() + { + public Long getValue() + { + return AutoRepair.instance.getRepairState(repairType).getTotalBytesToRepair(); + } + }); + bytesAlreadyRepaired = Metrics.register(factory.createMetricName("BytesAlreadyRepaired"), new Gauge() + { + public Long getValue() + { + return AutoRepair.instance.getRepairState(repairType).getBytesAlreadyRepaired(); + } + }); + totalKeyspaceRepairPlansToRepair = Metrics.register(factory.createMetricName("TotalKeyspaceRepairPlansToRepair"), new Gauge() + { + public Integer getValue() + { + return AutoRepair.instance.getRepairState(repairType).getTotalKeyspaceRepairPlansToRepair(); + } + }); + keyspaceRepairPlansAlreadyRepaired = Metrics.register(factory.createMetricName("KeyspaceRepairPlansAlreadyRepaired"), new Gauge() + { + public Integer getValue() + { + return AutoRepair.instance.getRepairState(repairType).getKeyspaceRepairPlansAlreadyRepaired(); + } + }); } public void recordTurn(AutoRepairUtils.RepairTurn turn) diff --git a/src/java/org/apache/cassandra/repair/autorepair/AutoRepair.java b/src/java/org/apache/cassandra/repair/autorepair/AutoRepair.java index 031b6352bf..0cfe2fa208 100644 --- a/src/java/org/apache/cassandra/repair/autorepair/AutoRepair.java +++ b/src/java/org/apache/cassandra/repair/autorepair/AutoRepair.java @@ -123,14 +123,15 @@ public class AutoRepair repairExecutors = new EnumMap<>(AutoRepairConfig.RepairType.class); repairRunnableExecutors = new EnumMap<>(AutoRepairConfig.RepairType.class); repairStates = new EnumMap<>(AutoRepairConfig.RepairType.class); + AutoRepairConfig config = DatabaseDescriptor.getAutoRepairConfig(); + for (AutoRepairConfig.RepairType repairType : AutoRepairConfig.RepairType.values()) { repairExecutors.put(repairType, executorFactory().scheduled(false, "AutoRepair-Repair-" + repairType.getConfigName(), Thread.NORM_PRIORITY)); repairRunnableExecutors.put(repairType, executorFactory().scheduled(false, "AutoRepair-RepairRunnable-" + repairType.getConfigName(), Thread.NORM_PRIORITY)); - repairStates.put(repairType, AutoRepairConfig.RepairType.getAutoRepairState(repairType)); + repairStates.put(repairType, AutoRepairConfig.RepairType.getAutoRepairState(repairType, config)); } - AutoRepairConfig config = DatabaseDescriptor.getAutoRepairConfig(); AutoRepairUtils.setup(); for (AutoRepairConfig.RepairType repairType : AutoRepairConfig.RepairType.values()) @@ -197,6 +198,8 @@ public class AutoRepair if (turn == MY_TURN || turn == MY_TURN_DUE_TO_PRIORITY || turn == MY_TURN_FORCE_REPAIR) { repairState.recordTurn(turn); + repairState.setBytesAlreadyRepaired(0L); + repairState.setKeyspaceRepairPlansAlreadyRepaired(0); // For normal auto repair, we will use primary range only repairs (Repair with -pr option). // For some cases, we may set the auto_repair_primary_token_range_only flag to false then we will do repair // without -pr. We may also do force repair for certain node that we want to repair all the data on one node @@ -231,23 +234,30 @@ public class AutoRepair } // Separate out the keyspaces and tables to repair based on their priority, with each repair plan representing a uniquely occuring priority. - List repairPlans = PrioritizedRepairPlan.build(keyspacesAndTablesToRepair, repairType, shuffleFunc); + List repairPlans = PrioritizedRepairPlan.build(keyspacesAndTablesToRepair, repairType, shuffleFunc, primaryRangeOnly); + repairState.updateRepairScheduleStatistics(repairPlans); // calculate the repair assignments for each priority:keyspace. Iterator repairAssignmentsIterator = config.getTokenRangeSplitterInstance(repairType).getRepairAssignments(primaryRangeOnly, repairPlans); + int keyspaceRepairAssignmentsAlreadyRepaired = 0; while (repairAssignmentsIterator.hasNext()) { KeyspaceRepairAssignments repairAssignments = repairAssignmentsIterator.next(); List assignments = repairAssignments.getRepairAssignments(); if (assignments.isEmpty()) { + keyspaceRepairAssignmentsAlreadyRepaired++; logger.info("Skipping repairs for priorityBucket={} for keyspace={} since it yielded no assignments", repairAssignments.getPriority(), repairAssignments.getKeyspaceName()); continue; } - logger.info("Submitting repairs for priorityBucket={} for keyspace={} with assignmentCount={}", repairAssignments.getPriority(), repairAssignments.getKeyspaceName(), repairAssignments.getRepairAssignments().size()); + logger.info("Submitting repairs for priorityBucket={} for keyspace={} with assignmentCount={} and keyspaceRepairAssignmentsAlreadyRepaired={}/{}", + repairAssignments.getPriority(), repairAssignments.getKeyspaceName(), repairAssignments.getRepairAssignments().size(), + keyspaceRepairAssignmentsAlreadyRepaired, repairState.getTotalKeyspaceRepairPlansToRepair()); repairKeyspace(repairType, primaryRangeOnly, repairAssignments.getKeyspaceName(), repairAssignments.getRepairAssignments(), collectedRepairStats); + keyspaceRepairAssignmentsAlreadyRepaired++; + repairState.setKeyspaceRepairPlansAlreadyRepaired(keyspaceRepairAssignmentsAlreadyRepaired); } cleanupAndUpdateStats(turn, repairType, repairState, myId, startTimeInMillis, collectedRepairStats); @@ -277,6 +287,7 @@ public class AutoRepair long tableStartTime = timeFunc.get(); int totalProcessedAssignments = 0; Set> ranges = new HashSet<>(); + long bytesAlreadyRepaired = repairState.getBytesAlreadyRepaired(); for (RepairAssignment curRepairAssignment : repairAssignments) { try @@ -380,7 +391,10 @@ public class AutoRepair } ranges.clear(); } - logger.info("Repair completed for {} tables {}, range {}", keyspaceName, curRepairAssignment.getTableNames(), curRepairAssignment.getTokenRange()); + bytesAlreadyRepaired += curRepairAssignment.getEstimatedBytes(); + repairState.setBytesAlreadyRepaired(bytesAlreadyRepaired); + logger.info("Repair completed for {} tables {}, range {}, bytesAlreadyRepaired {}/{}", + keyspaceName, curRepairAssignment.getTableNames(), curRepairAssignment.getTokenRange(), bytesAlreadyRepaired, repairState.getTotalBytesToRepair()); } catch (Exception e) { @@ -492,8 +506,8 @@ public class AutoRepair TimeUnit.SECONDS.toDays(repairState.getClusterRepairTimeInSec())); } repairState.setLastRepairTime(timeFunc.get()); - repairState.setRepairInProgress(false); + AutoRepairUtils.updateFinishAutoRepairHistory(repairType, myId, timeFunc.get()); } diff --git a/src/java/org/apache/cassandra/repair/autorepair/AutoRepairConfig.java b/src/java/org/apache/cassandra/repair/autorepair/AutoRepairConfig.java index 7285af7948..500cbc09be 100644 --- a/src/java/org/apache/cassandra/repair/autorepair/AutoRepairConfig.java +++ b/src/java/org/apache/cassandra/repair/autorepair/AutoRepairConfig.java @@ -92,16 +92,16 @@ public class AutoRepairConfig implements Serializable return configName; } - public static AutoRepairState getAutoRepairState(RepairType repairType) + public static AutoRepairState getAutoRepairState(RepairType repairType, AutoRepairConfig config) { switch (repairType) { case FULL: - return new FullRepairState(); + return new FullRepairState(config); case INCREMENTAL: - return new IncrementalRepairState(); + return new IncrementalRepairState(config); case PREVIEW_REPAIRED: - return new PreviewRepairedState(); + return new PreviewRepairedState(config); } throw new IllegalArgumentException("Invalid repair type: " + repairType); diff --git a/src/java/org/apache/cassandra/repair/autorepair/AutoRepairState.java b/src/java/org/apache/cassandra/repair/autorepair/AutoRepairState.java index 6822f20cf0..fe0f566c46 100644 --- a/src/java/org/apache/cassandra/repair/autorepair/AutoRepairState.java +++ b/src/java/org/apache/cassandra/repair/autorepair/AutoRepairState.java @@ -60,6 +60,8 @@ public abstract class AutoRepairState @VisibleForTesting protected final RepairType repairType; @VisibleForTesting + protected AutoRepairConfig config; + @VisibleForTesting protected int totalTablesConsideredForRepair = 0; @VisibleForTesting protected long lastRepairTimeInMs; @@ -84,13 +86,22 @@ public abstract class AutoRepairState @VisibleForTesting protected int skippedTablesCount = 0; @VisibleForTesting + protected long totalBytesToRepair = 0; + @VisibleForTesting + protected long bytesAlreadyRepaired = 0; + @VisibleForTesting + protected int totalKeyspaceRepairPlansToRepair = 0; + @VisibleForTesting + protected int keyspaceRepairPlansAlreadyRepaired = 0; + @VisibleForTesting protected AutoRepairHistory longestUnrepairedNode; protected final AutoRepairMetrics metrics; - protected AutoRepairState(RepairType repairType) + protected AutoRepairState(RepairType repairType, AutoRepairConfig config) { metrics = AutoRepairMetricsManager.getMetrics(repairType); this.repairType = repairType; + this.config = config; } public abstract RepairCoordinator getRepairRunnable(String keyspace, List tables, Set> ranges, boolean primaryRangeOnly); @@ -98,7 +109,15 @@ public abstract class AutoRepairState protected RepairCoordinator getRepairRunnable(String keyspace, RepairOption options) { return new RepairCoordinator(StorageService.instance, StorageService.nextRepairCommand.incrementAndGet(), - options, keyspace); + options, keyspace); + } + + public void updateRepairScheduleStatistics(List repairPlans) + { + setTotalBytesToRepair(repairPlans.stream(). + flatMap(repairPlan -> repairPlan.getKeyspaceRepairPlans(). + stream()).mapToLong(KeyspaceRepairPlan::getEstimatedBytes).sum()); + setTotalKeyspaceRepairPlansToRepair(repairPlans.stream().mapToInt(repairPlan -> repairPlan.getKeyspaceRepairPlans().size()).sum()); } public long getLastRepairTime() @@ -239,20 +258,60 @@ public abstract class AutoRepairState { return totalDisabledTablesRepairCount; } + + public void setTotalBytesToRepair(long totalBytesToRepair) + { + this.totalBytesToRepair = totalBytesToRepair; + } + + public long getTotalBytesToRepair() + { + return totalBytesToRepair; + } + + public void setBytesAlreadyRepaired(long bytesAlreadyRepaired) + { + this.bytesAlreadyRepaired = bytesAlreadyRepaired; + } + + public long getBytesAlreadyRepaired() + { + return bytesAlreadyRepaired; + } + + public void setTotalKeyspaceRepairPlansToRepair(int totalKeyspaceRepairPlansToRepair) + { + this.totalKeyspaceRepairPlansToRepair = totalKeyspaceRepairPlansToRepair; + } + + public int getTotalKeyspaceRepairPlansToRepair() + { + return totalKeyspaceRepairPlansToRepair; + } + + public void setKeyspaceRepairPlansAlreadyRepaired(int keyspaceRepairPlansAlreadyRepaired) + { + this.keyspaceRepairPlansAlreadyRepaired = keyspaceRepairPlansAlreadyRepaired; + } + + public int getKeyspaceRepairPlansAlreadyRepaired() + { + return keyspaceRepairPlansAlreadyRepaired; + } } class PreviewRepairedState extends AutoRepairState { - public PreviewRepairedState() + public PreviewRepairedState(AutoRepairConfig config) { - super(RepairType.PREVIEW_REPAIRED); + super(RepairType.PREVIEW_REPAIRED, config); } @Override public RepairCoordinator getRepairRunnable(String keyspace, List tables, Set> ranges, boolean primaryRangeOnly) { RepairOption option = new RepairOption(RepairParallelism.PARALLEL, primaryRangeOnly, false, false, - AutoRepairService.instance.getAutoRepairConfig().getRepairThreads(repairType), ranges, false, false, PreviewKind.REPAIRED, false, true, true, false, false, false); + AutoRepairService.instance.getAutoRepairConfig().getRepairThreads(repairType), ranges, false, false, PreviewKind.REPAIRED, false, true, true, false, false, false); option.getColumnFamilies().addAll(tables); @@ -262,9 +321,9 @@ class PreviewRepairedState extends AutoRepairState class IncrementalRepairState extends AutoRepairState { - public IncrementalRepairState() + public IncrementalRepairState(AutoRepairConfig config) { - super(RepairType.INCREMENTAL); + super(RepairType.INCREMENTAL, config); } @Override @@ -307,9 +366,9 @@ class IncrementalRepairState extends AutoRepairState class FullRepairState extends AutoRepairState { - public FullRepairState() + public FullRepairState(AutoRepairConfig config) { - super(RepairType.FULL); + super(RepairType.FULL, config); } @Override diff --git a/src/java/org/apache/cassandra/repair/autorepair/AutoRepairUtils.java b/src/java/org/apache/cassandra/repair/autorepair/AutoRepairUtils.java index d0da66c6ff..d8c0c52f20 100644 --- a/src/java/org/apache/cassandra/repair/autorepair/AutoRepairUtils.java +++ b/src/java/org/apache/cassandra/repair/autorepair/AutoRepairUtils.java @@ -17,6 +17,7 @@ */ package org.apache.cassandra.repair.autorepair; +import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -43,9 +44,15 @@ import com.google.common.base.MoreObjects; import com.google.common.collect.Lists; import com.google.common.collect.Sets; +import com.clearspring.analytics.stream.cardinality.CardinalityMergeException; +import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus; +import com.clearspring.analytics.stream.cardinality.ICardinality; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Splitter; import org.apache.cassandra.dht.Token; +import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.io.sstable.metadata.CompactionMetadata; +import org.apache.cassandra.io.sstable.metadata.MetadataType; import org.apache.cassandra.locator.EndpointsByRange; import org.apache.cassandra.locator.EndpointsForRange; import org.apache.cassandra.locator.LocalStrategy; @@ -83,6 +90,7 @@ import org.apache.cassandra.service.ClientState; import org.apache.cassandra.service.QueryState; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.compatibility.TokenRingUtils; import org.apache.cassandra.tcm.membership.Directory; import org.apache.cassandra.tcm.membership.NodeAddresses; import org.apache.cassandra.tcm.membership.NodeId; @@ -93,6 +101,7 @@ import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.repair.autorepair.AutoRepairConfig.RepairType; import org.apache.cassandra.utils.NoSpamLogger; +import org.apache.cassandra.utils.concurrent.Refs; import static org.apache.cassandra.repair.autorepair.AutoRepairUtils.RepairTurn.MY_TURN; import static org.apache.cassandra.repair.autorepair.AutoRepairUtils.RepairTurn.MY_TURN_DUE_TO_PRIORITY; @@ -312,7 +321,7 @@ public class AutoRepairUtils public Set getAllHostsWithOngoingRepair() { - return Sets.union(hostIdsWithOnGoingRepair, hostIdsWithOnGoingForceRepair); + return Sets.union(hostIdsWithOnGoingRepair, hostIdsWithOnGoingForceRepair); } public String toString() @@ -484,6 +493,7 @@ public class AutoRepairUtils /** * Convenience method to resolve the broadcast address of a host id from {@link ClusterMetadata} + * * @return broadcast address if it exists in CMS, otherwise null. */ @Nullable @@ -551,7 +561,7 @@ public class AutoRepairUtils * Accepts the currently evaluated repairType's schedule as an optimization to avoid grabbing its repair status an * additional time. * - * @param myRepairType The repair type schedule being evaluated. + * @param myRepairType The repair type schedule being evaluated. * @param myRepairStatus The repair status for that repair type. * @return All hosts among active schedules currently being repaired. */ @@ -591,6 +601,7 @@ public class AutoRepairUtils * Identifies the most eligible host to repair for nodes preceding or equal to this nodes' lastRepairFinishTime. * The criteria for this is to find the node with the oldest last repair finish time of which none of its replicas * are currently under repair. + * * @return The most eligible host to repair or null if no candidates before and including this nodes' current repair status. */ @VisibleForTesting @@ -666,13 +677,13 @@ public class AutoRepairUtils /** - * @return Whether the host for the given eligibleRepairHistory has any replicas in hostsBeingRepaired. - * @param eligibleHistory History of node to check - * @param myId Host id of this node, if the repair history is for this node, additional logging will take place. - * @param myRepairType repair type being evaluated - * @param hostsBeingRepaired Hosts being repaired. - * @param hostIdToRepairType mapping of hosts being repaired to the repair type its being repaired for. + * @param eligibleHistory History of node to check + * @param myId Host id of this node, if the repair history is for this node, additional logging will take place. + * @param myRepairType repair type being evaluated + * @param hostsBeingRepaired Hosts being repaired. + * @param hostIdToRepairType mapping of hosts being repaired to the repair type its being repaired for. * @param replicationStrategies Mapping of unique replication strategies to keyspaces having that strategy. + * @return Whether the host for the given eligibleRepairHistory has any replicas in hostsBeingRepaired. */ private static boolean hasReplicaWithOngoingRepair(AutoRepairHistory eligibleHistory, UUID myId, @@ -726,7 +737,6 @@ public class AutoRepairUtils eligibleHistory.hostId, eligibleBroadcastAddress, hostId, inetAddressAndPort, entry.getValue().size(), entry.getValue().get(0), hostIdToRepairType.get(hostId)); - } return true; } @@ -871,7 +881,7 @@ public class AutoRepairUtils } int parallelRepairNumber = getMaxNumberOfNodeRunAutoRepair(repairType, - autoRepairHistories == null ? 0 : autoRepairHistories.size()); + autoRepairHistories == null ? 0 : autoRepairHistories.size()); logger.info("Will run repairs concurrently on {} node(s)", parallelRepairNumber); if (currentRepairStatus == null || parallelRepairNumber > currentRepairStatus.hostIdsWithOnGoingRepair.size()) { @@ -937,8 +947,8 @@ public class AutoRepairUtils // log which node is next, which is helpful for debugging logger.info("Next node to be repaired for repair type {}: {} ({})", repairType, - getBroadcastAddress(nodeToBeRepaired.hostId), - nodeToBeRepaired); + getBroadcastAddress(nodeToBeRepaired.hostId), + nodeToBeRepaired); } // If this node is not identified as most eligible, set the repair lag time. @@ -1201,4 +1211,193 @@ public class AutoRepairUtils } return ranges; } + + /** + * Finds a list of SSTables for a given {@code repairType}, + * {@code keyspace}, {@code table}, and {@code tokenRange} and then it internally calls + * another API {@code AutoRepairUtils.getSizesForRangeOfSSTables}, which figures out the estimated data size. + * + * @param repairType the repair type (e.g., FULL, INCREMENTAL) + * @param keyspace the keyspace name + * @param table the table name + * @param tokenRange the token range to evaluate + * @return an estimate representing the number of partitions, size in range, and total size + */ + static SizeEstimate getRangeSizeEstimate(RepairType repairType, String keyspace, String table, Range tokenRange) + { + logger.debug("Calculating size estimate for {}.{} for range {}", keyspace, table, tokenRange); + try (Refs refs = RepairTokenRangeSplitter.getSSTableReaderRefs(repairType, keyspace, table, tokenRange)) + { + SizeEstimate estimate = getSizesForRangeOfSSTables(repairType, keyspace, table, tokenRange, refs); + logger.debug("Generated size estimate {}", estimate); + return estimate; + } + } + /** + * Calculates the size estimation qualified to be repaired for a given {@code repairType}, + * {@code keyspace}, {@code table}, {@code tokenRange}, and {@code refs}. + *

+ * If the compression is enabled, then the size will be an estimate, otherwise it will be accurate. + *

+ * + * @param repairType + * @param keyspace + * @param table + * @param tokenRange + * @param refs + * @return an estimate representing the number of partitions, size in range, and total size + */ + static SizeEstimate getSizesForRangeOfSSTables(RepairType repairType, String keyspace, String table, + Range tokenRange, Refs refs) + { + List> singletonRange = Collections.singletonList(tokenRange); + ICardinality cardinality = new HyperLogLogPlus(13, 25); + long approxBytesInRange = 0L; + long totalBytes = 0L; + + for (SSTableReader reader : refs) + { + try + { + if (reader.openReason == SSTableReader.OpenReason.EARLY) + continue; + CompactionMetadata metadata = (CompactionMetadata) reader.descriptor.getMetadataSerializer().deserialize(reader.descriptor, MetadataType.COMPACTION); + if (metadata != null) + cardinality = cardinality.merge(metadata.cardinalityEstimator); + + // use onDiskLength, which is the actual size of the SSTable data file. + long sstableSize = reader.onDiskLength(); + totalBytes += sstableSize; + + // get the on disk size for the token range, note for compressed data this includes the full + // chunks the start and end ranges are found in. + long approximateRangeBytesInSSTable = reader.onDiskSizeForPartitionPositions(reader.getPositionsForRanges(singletonRange)); + approxBytesInRange += Math.min(approximateRangeBytesInSSTable, sstableSize); + } + catch (IOException | CardinalityMergeException e) + { + logger.error("Error calculating size estimate for {}.{} for range {} on {}", keyspace, table, tokenRange, reader, e); + } + } + + long partitions = 0L; + if (totalBytes > 0) + { + // use the ratio from size to estimate the partitions in the range as well + double ratio = approxBytesInRange / (double) totalBytes; + partitions = (long) Math.max(1, Math.ceil(cardinality.cardinality() * ratio)); + } + return new SizeEstimate(repairType, keyspace, table, tokenRange, partitions, approxBytesInRange, totalBytes); + } + + /** + * Calculates the token ranges owned by this node for a given keyspace. + * + * @param primaryRangeOnly whether to use only primary token ranges or include replicated ones + * @param keyspaceName the name of the keyspace + * @return one or more token ranges owned by this node + */ + static List> getTokenRanges(boolean primaryRangeOnly, String keyspaceName) + { + // Collect all applicable token ranges + Collection> wrappedRanges; + if (primaryRangeOnly) + { + wrappedRanges = TokenRingUtils.getPrimaryRangesForEndpoint(keyspaceName, FBUtilities.getBroadcastAddressAndPort()); + } + else + { + wrappedRanges = StorageService.instance.getLocalRanges(keyspaceName); + } + + // Unwrap each range as we need to account for ranges that overlap the ring + List> ranges = new ArrayList<>(); + for (Range wrappedRange : wrappedRanges) + { + ranges.addAll(wrappedRange.unwrap()); + } + return ranges; + } + + /** + * Calculates the total bytes to be repaired for a given keyspace and list of tables. + * + * @param repairType the repair type (e.g., FULL, INCREMENTAL) + * @param keyspaceName the name of the keyspace + * @param tableNames the list of tables + * @return a key-value map where the key is {@code keyspaceName.tableName} and the value is the number of bytes + * to be repaired. + */ + public static Map, SizeEstimate>> calcTotalBytesToBeRepaired(RepairType repairType, String keyspaceName, List tableNames, List> tokenRanges) + { + Map, SizeEstimate>> ksTablesEstimatedBytes = new HashMap<>(); + for (String tableName : tableNames) + { + String ksTable = getKeyspaceTableName(keyspaceName, tableName); + ksTablesEstimatedBytes.computeIfAbsent(ksTable, k -> new HashMap<>()); + Map, SizeEstimate> tokenToSize = ksTablesEstimatedBytes.get(ksTable); + for (Range tokenRange : tokenRanges) + { + SizeEstimate tableAssignments = getRangeSizeEstimate(repairType, keyspaceName, tableName, tokenRange); + tokenToSize.put(tokenRange, tableAssignments); + } + } + return ksTablesEstimatedBytes; + } + + public static String getKeyspaceTableName(String keyspace, String table) + { + return keyspace + "." + table; + } + + /** + * Represents a size estimate by both bytes and partition count for a given keyspace and table for a token range. + */ + @VisibleForTesting + protected static class SizeEstimate + { + public final RepairType repairType; + public final String keyspace; + public final String table; + public final Range tokenRange; + public final long partitions; + public final long sizeInRange; + public final long totalSize; + /** + * Size to consider in the repair. For incremental repair, we want to consider the total size + * of the estimate as we have to factor in anticompacting the entire SSTable. + * For full repair, just use the size containing the range. + */ + public final long sizeForRepair; + + public SizeEstimate(RepairType repairType, + String keyspace, String table, Range tokenRange, + long partitions, long sizeInRange, long totalSize) + { + this.repairType = repairType; + this.keyspace = keyspace; + this.table = table; + this.tokenRange = tokenRange; + this.partitions = partitions; + this.sizeInRange = sizeInRange; + this.totalSize = totalSize; + + this.sizeForRepair = repairType == RepairType.INCREMENTAL ? totalSize : sizeInRange; + } + + @Override + public String toString() + { + return "SizeEstimate{" + + "repairType=" + repairType + + ", keyspace='" + keyspace + '\'' + + ", table='" + table + '\'' + + ", tokenRange=" + tokenRange + + ", partitions=" + partitions + + ", sizeInRange=" + sizeInRange + + ", totalSize=" + totalSize + + ", sizeForRepair=" + sizeForRepair + + '}'; + } + } } diff --git a/src/java/org/apache/cassandra/repair/autorepair/FixedSplitTokenRangeSplitter.java b/src/java/org/apache/cassandra/repair/autorepair/FixedSplitTokenRangeSplitter.java index a6dddb3060..4dd1e409c8 100644 --- a/src/java/org/apache/cassandra/repair/autorepair/FixedSplitTokenRangeSplitter.java +++ b/src/java/org/apache/cassandra/repair/autorepair/FixedSplitTokenRangeSplitter.java @@ -32,10 +32,6 @@ import org.apache.cassandra.service.AutoRepairService; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; -import org.apache.cassandra.service.StorageService; -import org.apache.cassandra.tcm.compatibility.TokenRingUtils; -import org.apache.cassandra.utils.FBUtilities; - import static org.apache.cassandra.repair.autorepair.AutoRepairUtils.split; /** @@ -98,13 +94,7 @@ public class FixedSplitTokenRangeSplitter implements IAutoRepairTokenRangeSplitt String keyspaceName = repairPlan.getKeyspaceName(); List tableNames = repairPlan.getTableNames(); - Collection> tokens = TokenRingUtils.getPrimaryRangesForEndpoint(keyspaceName, FBUtilities.getBroadcastAddressAndPort()); - if (!primaryRangeOnly) - { - // if we need to repair non-primary token ranges, then change the tokens accordingly - tokens = StorageService.instance.getLocalReplicas(keyspaceName).onlyFull().ranges(); - } - + Collection> tokens = AutoRepairUtils.getTokenRanges(primaryRangeOnly, keyspaceName); boolean byKeyspace = config.getRepairByKeyspace(repairType); // collect all token ranges. List> allRanges = new ArrayList<>(); @@ -117,10 +107,15 @@ public class FixedSplitTokenRangeSplitter implements IAutoRepairTokenRangeSplitt if (byKeyspace) { + + // This calculation is the best effort for the FixedSplitTokenRangeSplitter. + // In practice, this metric may not give you an accurate view in case of uneven data distribution. + long totalBytes = repairPlan.getEstimatedBytes(); + long bytesPerRange = Math.max(1, totalBytes / splitsPerRange); for (Range splitRange : allRanges) { // add repair assignment for each range entire keyspace's tables - repairAssignments.add(new RepairAssignment(splitRange, keyspaceName, tableNames)); + repairAssignments.add(new RepairAssignment(splitRange, keyspaceName, tableNames, bytesPerRange)); } } else @@ -128,9 +123,11 @@ public class FixedSplitTokenRangeSplitter implements IAutoRepairTokenRangeSplitt // add repair assignment per table for (String tableName : tableNames) { + long totalBytes = repairPlan.getTableEstimatedBytes(AutoRepairUtils.getKeyspaceTableName(keyspaceName, tableName)); + long bytesPerRange = Math.max(1, totalBytes / splitsPerRange); for (Range splitRange : allRanges) { - repairAssignments.add(new RepairAssignment(splitRange, keyspaceName, Collections.singletonList(tableName))); + repairAssignments.add(new RepairAssignment(splitRange, keyspaceName, Collections.singletonList(tableName), bytesPerRange)); } } } diff --git a/src/java/org/apache/cassandra/repair/autorepair/KeyspaceRepairPlan.java b/src/java/org/apache/cassandra/repair/autorepair/KeyspaceRepairPlan.java index 3c13e3d80d..750e56e9d2 100644 --- a/src/java/org/apache/cassandra/repair/autorepair/KeyspaceRepairPlan.java +++ b/src/java/org/apache/cassandra/repair/autorepair/KeyspaceRepairPlan.java @@ -18,9 +18,16 @@ package org.apache.cassandra.repair.autorepair; +import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.Objects; +import com.google.common.annotations.VisibleForTesting; + +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.Token; + /** * Encapsulates an intent to repair the given keyspace's tables */ @@ -30,10 +37,14 @@ public class KeyspaceRepairPlan private final List tableNames; - public KeyspaceRepairPlan(String keyspaceName, List tableNames) + @VisibleForTesting + public Map, AutoRepairUtils.SizeEstimate>> ksTablesEstimatedBytes; + + public KeyspaceRepairPlan(String keyspaceName, List tableNames, Map, AutoRepairUtils.SizeEstimate>> ksTablesEstimatedBytes) { this.keyspaceName = keyspaceName; this.tableNames = tableNames; + this.ksTablesEstimatedBytes = ksTablesEstimatedBytes; } public String getKeyspaceName() @@ -46,18 +57,40 @@ public class KeyspaceRepairPlan return tableNames; } + public long getEstimatedBytes() + { + return ksTablesEstimatedBytes.values().stream() + .flatMap(tableMap -> tableMap.values().stream()) + .mapToLong(sizeEstimate -> sizeEstimate.sizeForRepair) + .sum(); + } + + public long getTableEstimatedBytes(String keyspaceTableName) + { + return ksTablesEstimatedBytes.getOrDefault(keyspaceTableName, + Collections.emptyMap()).values().stream().mapToLong(sizeEstimate -> sizeEstimate.sizeForRepair).sum(); + } + + public AutoRepairUtils.SizeEstimate getSizeEstimate(String keyspaceTableName, Range tokenRange) + { + return ksTablesEstimatedBytes == null ? null + : ksTablesEstimatedBytes.getOrDefault(keyspaceTableName, null) == null ? null + : ksTablesEstimatedBytes.get(keyspaceTableName).get(tokenRange); + } + @Override public boolean equals(Object o) { if (o == null || getClass() != o.getClass()) return false; KeyspaceRepairPlan that = (KeyspaceRepairPlan) o; - return Objects.equals(keyspaceName, that.keyspaceName) && Objects.equals(tableNames, that.tableNames); + return Objects.equals(keyspaceName, that.keyspaceName) && Objects.equals(tableNames, that.tableNames) + && Objects.equals(ksTablesEstimatedBytes, that.ksTablesEstimatedBytes); } @Override public int hashCode() { - return Objects.hash(keyspaceName, tableNames); + return Objects.hash(keyspaceName, tableNames, ksTablesEstimatedBytes); } @Override @@ -66,6 +99,7 @@ public class KeyspaceRepairPlan return "KeyspaceRepairPlan{" + "keyspaceName='" + keyspaceName + '\'' + ", tableNames=" + tableNames + + ", ksTablesEstimatedBytes=" + ksTablesEstimatedBytes + '}'; } } diff --git a/src/java/org/apache/cassandra/repair/autorepair/PrioritizedRepairPlan.java b/src/java/org/apache/cassandra/repair/autorepair/PrioritizedRepairPlan.java index fbedb71b7c..4457ccdd34 100644 --- a/src/java/org/apache/cassandra/repair/autorepair/PrioritizedRepairPlan.java +++ b/src/java/org/apache/cassandra/repair/autorepair/PrioritizedRepairPlan.java @@ -19,7 +19,6 @@ package org.apache.cassandra.repair.autorepair; import java.util.ArrayList; -import java.util.Arrays; import java.util.Comparator; import java.util.HashMap; import java.util.List; @@ -89,11 +88,11 @@ public class PrioritizedRepairPlan * for their given priority. * * @param keyspacesToTableNames A mapping keyspace to table names - * @param repairType The repair type that is being executed - * @param orderFunc A function to order keyspace and tables in the returned plan. + * @param repairType The repair type that is being executed + * @param orderFunc A function to order keyspace and tables in the returned plan. * @return Ordered list of plan's by table priorities. */ - public static List build(Map> keyspacesToTableNames, AutoRepairConfig.RepairType repairType, Consumer> orderFunc) + public static List build(Map> keyspacesToTableNames, AutoRepairConfig.RepairType repairType, Consumer> orderFunc, boolean primaryRangeOnly) { // Build a map of priority -> (keyspace -> tables) Map>> plans = new HashMap<>(); @@ -124,31 +123,20 @@ public class PrioritizedRepairPlan List keyspaceNames = new ArrayList<>(keyspacesAndTables.keySet()); orderFunc.accept(keyspaceNames); - for(String keyspaceName : keyspaceNames) + for (String keyspaceName : keyspaceNames) { - List tableNames = keyspacesAndTables.get(keyspaceName); - orderFunc.accept(tableNames); - KeyspaceRepairPlan keyspaceRepairPlan = new KeyspaceRepairPlan(keyspaceName, new ArrayList<>(tableNames)); - keyspaceRepairPlans.add(keyspaceRepairPlan); + List tableNames = keyspacesAndTables.get(keyspaceName); + orderFunc.accept(tableNames); + KeyspaceRepairPlan keyspaceRepairPlan = + new KeyspaceRepairPlan(keyspaceName, new ArrayList<>(tableNames), + AutoRepairUtils.calcTotalBytesToBeRepaired(repairType, keyspaceName, tableNames, AutoRepairUtils.getTokenRanges(primaryRangeOnly, keyspaceName))); + keyspaceRepairPlans.add(keyspaceRepairPlan); } } return planList; } - /** - * Convenience method to build a repair plan for a single keyspace with tables. Primarily useful in testing. - * @param keyspaceName Keyspace to repair - * @param tableNames tables to repair for the given keyspace. - * @return Single repair plan. - */ - static List buildSingleKeyspacePlan(AutoRepairConfig.RepairType repairType, String keyspaceName, String ... tableNames) - { - Map> keyspaceMap = new HashMap<>(); - keyspaceMap.put(keyspaceName, Arrays.asList(tableNames)); - return build(keyspaceMap, repairType, (l) -> {}); - } - /** * @return The priority of the given table if defined, otherwise 0. */ diff --git a/src/java/org/apache/cassandra/repair/autorepair/RepairAssignment.java b/src/java/org/apache/cassandra/repair/autorepair/RepairAssignment.java index 63f8fbed44..6e07399aad 100644 --- a/src/java/org/apache/cassandra/repair/autorepair/RepairAssignment.java +++ b/src/java/org/apache/cassandra/repair/autorepair/RepairAssignment.java @@ -35,11 +35,14 @@ public class RepairAssignment final List tableNames; - public RepairAssignment(Range tokenRange, String keyspaceName, List tableNames) + protected final long estimatedBytes; + + public RepairAssignment(Range tokenRange, String keyspaceName, List tableNames, long estimatedBytes) { this.tokenRange = tokenRange; this.keyspaceName = keyspaceName; this.tableNames = tableNames; + this.estimatedBytes = estimatedBytes; } public Range getTokenRange() @@ -57,19 +60,25 @@ public class RepairAssignment return tableNames; } + public long getEstimatedBytes() + { + return estimatedBytes; + } + @Override public boolean equals(Object o) { if (this == o) return true; if (o == null || getClass() != o.getClass()) return false; RepairAssignment that = (RepairAssignment) o; - return Objects.equals(tokenRange, that.tokenRange) && Objects.equals(keyspaceName, that.keyspaceName) && Objects.equals(tableNames, that.tableNames); + return Objects.equals(tokenRange, that.tokenRange) && Objects.equals(keyspaceName, that.keyspaceName) + && Objects.equals(tableNames, that.tableNames) && Objects.equals(estimatedBytes, that.estimatedBytes); } @Override public int hashCode() { - return Objects.hash(tokenRange, keyspaceName, tableNames); + return Objects.hash(tokenRange, keyspaceName, tableNames, estimatedBytes); } @Override @@ -79,6 +88,7 @@ public class RepairAssignment "tokenRange=" + tokenRange + ", keyspaceName='" + keyspaceName + '\'' + ", tableNames=" + tableNames + + ", estimatedBytes=" + estimatedBytes + '}'; } } diff --git a/src/java/org/apache/cassandra/repair/autorepair/RepairTokenRangeSplitter.java b/src/java/org/apache/cassandra/repair/autorepair/RepairTokenRangeSplitter.java index 20a79adc8e..4b03f22604 100644 --- a/src/java/org/apache/cassandra/repair/autorepair/RepairTokenRangeSplitter.java +++ b/src/java/org/apache/cassandra/repair/autorepair/RepairTokenRangeSplitter.java @@ -18,7 +18,6 @@ package org.apache.cassandra.repair.autorepair; -import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -39,14 +38,9 @@ import java.util.stream.Collectors; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.ImmutableList; -import org.apache.cassandra.tcm.compatibility.TokenRingUtils; -import org.apache.cassandra.utils.FBUtilities; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.clearspring.analytics.stream.cardinality.CardinalityMergeException; -import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus; -import com.clearspring.analytics.stream.cardinality.ICardinality; import org.apache.cassandra.config.DataStorageSpec; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.PartitionPosition; @@ -56,11 +50,8 @@ import org.apache.cassandra.db.lifecycle.View; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.io.sstable.format.SSTableReader; -import org.apache.cassandra.io.sstable.metadata.CompactionMetadata; -import org.apache.cassandra.io.sstable.metadata.MetadataType; import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.service.AutoRepairService; -import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.concurrent.Refs; import static org.apache.cassandra.repair.autorepair.AutoRepairUtils.split; @@ -228,7 +219,8 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter * A custom {@link RepairAssignmentIterator} that confines the number of repair assignments to * max_bytes_per_schedule. */ - private class BytesBasedRepairAssignmentIterator extends RepairAssignmentIterator { + private class BytesBasedRepairAssignmentIterator extends RepairAssignmentIterator + { private final boolean primaryRangeOnly; private long bytesSoFar = 0; @@ -250,14 +242,14 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter return new KeyspaceRepairAssignments(priority, repairPlan.getKeyspaceName(), Collections.emptyList()); } - List> tokenRanges = getTokenRanges(primaryRangeOnly, repairPlan.getKeyspaceName()); + List> tokenRanges = AutoRepairUtils.getTokenRanges(primaryRangeOnly, repairPlan.getKeyspaceName()); // shuffle token ranges to unbias selection of ranges Collections.shuffle(tokenRanges); List repairAssignments = new ArrayList<>(); // Generate assignments for each range speparately for (Range tokenRange : tokenRanges) { - repairAssignments.addAll(getRepairAssignmentsForKeyspace(repairType, repairPlan.getKeyspaceName(), repairPlan.getTableNames(), tokenRange)); + repairAssignments.addAll(getRepairAssignmentsForKeyspace(repairType, repairPlan, tokenRange)); } FilteredRepairAssignments filteredRepairAssignments = filterRepairAssignments(priority, repairPlan.getKeyspaceName(), repairAssignments, bytesSoFar); @@ -267,7 +259,7 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter } @VisibleForTesting - List getRepairAssignmentsForKeyspace(AutoRepairConfig.RepairType repairType, String keyspaceName, List tableNames, Range tokenRange) + List getRepairAssignmentsForKeyspace(AutoRepairConfig.RepairType repairType, KeyspaceRepairPlan repairPlan, Range tokenRange) { List repairAssignments = new ArrayList<>(); // this is used for batching minimal single assignment tables together @@ -277,12 +269,12 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter // If we can repair by keyspace, sort the tables by size so can batch the smallest ones together boolean repairByKeyspace = config.getRepairByKeyspace(repairType); - List tablesToProcess = tableNames; + List tablesToProcess = repairPlan.getTableNames(); if (repairByKeyspace) { - tablesToProcess = tableNames.stream().sorted((t1, t2) -> { - ColumnFamilyStore cfs1 = ColumnFamilyStore.getIfExists(keyspaceName, t1); - ColumnFamilyStore cfs2 = ColumnFamilyStore.getIfExists(keyspaceName, t2); + tablesToProcess = repairPlan.getTableNames().stream().sorted((t1, t2) -> { + ColumnFamilyStore cfs1 = ColumnFamilyStore.getIfExists(repairPlan.getKeyspaceName(), t1); + ColumnFamilyStore cfs2 = ColumnFamilyStore.getIfExists(repairPlan.getKeyspaceName(), t2); // If for whatever reason the CFS is not retrievable, we can assume it has been deleted, so give the // other cfs precedence. if (cfs1 == null) @@ -301,7 +293,7 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter for (String tableName : tablesToProcess) { - List tableAssignments = getRepairAssignmentsForTable(keyspaceName, tableName, tokenRange); + List tableAssignments = getRepairAssignmentsForTable(repairPlan, tableName, tokenRange); if (tableAssignments.isEmpty()) continue; @@ -348,8 +340,9 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter /** * Given a repair type and map of sized-based repair assignments, confine them by maxBytesPerSchedule. + * * @param repairAssignments the assignments to filter. - * @param bytesSoFar repair assignment bytes accumulated so far. + * @param bytesSoFar repair assignment bytes accumulated so far. * @return A list of repair assignments confined by maxBytesPerSchedule. */ @VisibleForTesting @@ -444,9 +437,9 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter } /** + * @param repairAssignments The assignments to sum * @return The sum of {@link SizedRepairAssignment#getEstimatedBytes()} of all given * repairAssignments. - * @param repairAssignments The assignments to sum */ @VisibleForTesting protected static long getEstimatedBytes(List repairAssignments) @@ -484,10 +477,18 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter } @VisibleForTesting - protected List getRepairAssignmentsForTable(String keyspaceName, String tableName, Range tokenRange) + protected List getRepairAssignmentsForTable(KeyspaceRepairPlan repairPlan, String tableName, Range tokenRange) { - List sizeEstimates = getRangeSizeEstimate(keyspaceName, tableName, tokenRange); - return getRepairAssignments(sizeEstimates); + AutoRepairUtils.SizeEstimate sizeEstimate = repairPlan.getSizeEstimate(AutoRepairUtils.getKeyspaceTableName(repairPlan.getKeyspaceName(), tableName), tokenRange); + if (sizeEstimate == null) + { + // Ideally, it should have been cached already inside the KeyspaceRepairPlan, but incase it was not, + // then recalculating it. It is a bit expensive, but necessary for the repair + logger.warn("The size estimate for {}.{} range {} was not pre-calculated, calculating on-demand", + repairPlan.getKeyspaceName(), tableName, tokenRange); + sizeEstimate = AutoRepairUtils.getRangeSizeEstimate(repairType, repairPlan.getKeyspaceName(), tableName, tokenRange); + } + return getRepairAssignments(sizeEstimate); } private static void logSkippingTable(String keyspaceName, String tableName) @@ -496,82 +497,75 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter } @VisibleForTesting - protected List getRepairAssignments(List sizeEstimates) + protected List getRepairAssignments(AutoRepairUtils.SizeEstimate estimate) { List repairAssignments = new ArrayList<>(); // since its possible for us to hit maxBytesPerSchedule before seeing all ranges, shuffle so there is chance // at least of hitting all the ranges _eventually_ for the worst case scenarios - Collections.shuffle(sizeEstimates); int totalExpectedSubRanges = 0; - for (SizeEstimate estimate : sizeEstimates) + if (estimate.sizeForRepair != 0) { - if (estimate.sizeForRepair != 0) + boolean needsSplitting = estimate.sizeForRepair > bytesPerAssignment.toBytes() || estimate.partitions > partitionsPerAssignment; + if (needsSplitting) { - boolean needsSplitting = estimate.sizeForRepair > bytesPerAssignment.toBytes() || estimate.partitions > partitionsPerAssignment; - if (needsSplitting) - { - totalExpectedSubRanges += calculateNumberOfSplits(estimate); - } + totalExpectedSubRanges += calculateNumberOfSplits(estimate); } } - for (SizeEstimate estimate : sizeEstimates) + if (estimate.sizeForRepair == 0) { - if (estimate.sizeForRepair == 0) + ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(estimate.keyspace, estimate.table); + + if (cfs == null) { - ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(estimate.keyspace, estimate.table); + logSkippingTable(estimate.keyspace, estimate.table); + return Collections.emptyList(); + } - if (cfs == null) + long memtableSize = cfs.getTracker().getView().getCurrentMemtable().getLiveDataSize(); + if (memtableSize > 0L) + { + logger.debug("Included {}.{} range {}, had no unrepaired SSTables, but memtableSize={}, adding single repair assignment", estimate.keyspace, estimate.table, estimate.tokenRange, memtableSize); + SizedRepairAssignment assignment = new SizedRepairAssignment(estimate.tokenRange, estimate.keyspace, Collections.singletonList(estimate.table), "full primary rangee for table with memtable only detected", memtableSize); + repairAssignments.add(assignment); + } + else + { + logger.debug("Included {}.{} range {}, has no SSTables or memtable data, but adding single repair assignment for entire range in case writes were missed", estimate.keyspace, estimate.table, estimate.tokenRange); + SizedRepairAssignment assignment = new SizedRepairAssignment(estimate.tokenRange, estimate.keyspace, Collections.singletonList(estimate.table), "full primary range for table with no data detected", 0L); + repairAssignments.add(assignment); + } + } + else + { + // Check if the estimate needs splitting based on the criteria + boolean needsSplitting = estimate.sizeForRepair > bytesPerAssignment.toBytes() || estimate.partitions > partitionsPerAssignment; + if (needsSplitting) + { + int numberOfSplits = calculateNumberOfSplits(estimate); + long approximateBytesPerSplit = estimate.sizeForRepair / numberOfSplits; + Collection> subranges = split(estimate.tokenRange, numberOfSplits); + for (Range subrange : subranges) { - logSkippingTable(estimate.keyspace, estimate.table); - continue; - } - - long memtableSize = cfs.getTracker().getView().getCurrentMemtable().getLiveDataSize(); - if (memtableSize > 0L) - { - logger.debug("Included {}.{} range {}, had no unrepaired SSTables, but memtableSize={}, adding single repair assignment", estimate.keyspace, estimate.table, estimate.tokenRange, memtableSize); - SizedRepairAssignment assignment = new SizedRepairAssignment(estimate.tokenRange, estimate.keyspace, Collections.singletonList(estimate.table), "full primary rangee for table with memtable only detected", memtableSize); - repairAssignments.add(assignment); - } - else - { - logger.debug("Included {}.{} range {}, has no SSTables or memtable data, but adding single repair assignment for entire range in case writes were missed", estimate.keyspace, estimate.table, estimate.tokenRange); - SizedRepairAssignment assignment = new SizedRepairAssignment(estimate.tokenRange, estimate.keyspace, Collections.singletonList(estimate.table), "full primary range for table with no data detected", 0L); + SizedRepairAssignment assignment = new SizedRepairAssignment(subrange, estimate.keyspace, Collections.singletonList(estimate.table), + String.format("subrange %d of %d", repairAssignments.size() + 1, totalExpectedSubRanges), + approximateBytesPerSplit); repairAssignments.add(assignment); } } else { - // Check if the estimate needs splitting based on the criteria - boolean needsSplitting = estimate.sizeForRepair > bytesPerAssignment.toBytes() || estimate.partitions > partitionsPerAssignment; - if (needsSplitting) - { - int numberOfSplits = calculateNumberOfSplits(estimate); - long approximateBytesPerSplit = estimate.sizeForRepair / numberOfSplits; - Collection> subranges = split(estimate.tokenRange, numberOfSplits); - for (Range subrange : subranges) - { - SizedRepairAssignment assignment = new SizedRepairAssignment(subrange, estimate.keyspace, Collections.singletonList(estimate.table), - String.format("subrange %d of %d", repairAssignments.size()+1, totalExpectedSubRanges), - approximateBytesPerSplit); - repairAssignments.add(assignment); - } - } - else - { - // No splitting needed, repair the entire range as-is - SizedRepairAssignment assignment = new SizedRepairAssignment(estimate.tokenRange, estimate.keyspace, - Collections.singletonList(estimate.table), - "full primary range for table", estimate.sizeForRepair); - repairAssignments.add(assignment); - } + // No splitting needed, repair the entire range as-is + SizedRepairAssignment assignment = new SizedRepairAssignment(estimate.tokenRange, estimate.keyspace, + Collections.singletonList(estimate.table), + "full primary range for table", estimate.sizeForRepair); + repairAssignments.add(assignment); } } return repairAssignments; } - private int calculateNumberOfSplits(SizeEstimate estimate) + private int calculateNumberOfSplits(AutoRepairUtils.SizeEstimate estimate) { // Calculate the number of splits needed for size and partitions int splitsForSize = (int) Math.ceil((double) estimate.sizeForRepair / bytesPerAssignment.toBytes()); @@ -597,84 +591,6 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter return splits; } - private List> getTokenRanges(boolean primaryRangeOnly, String keyspaceName) - { - // Collect all applicable token ranges - Collection> wrappedRanges; - if (primaryRangeOnly) - { - wrappedRanges = TokenRingUtils.getPrimaryRangesForEndpoint(keyspaceName, FBUtilities.getBroadcastAddressAndPort()); - } - else - { - wrappedRanges = StorageService.instance.getLocalRanges(keyspaceName); - } - - // Unwrap each range as we need to account for ranges that overlap the ring - List> ranges = new ArrayList<>(); - for (Range wrappedRange : wrappedRanges) - { - ranges.addAll(wrappedRange.unwrap()); - } - return ranges; - } - - private List getRangeSizeEstimate(String keyspace, String table, Range tokenRange) - { - List sizeEstimates = new ArrayList<>(); - logger.debug("Calculating size estimate for {}.{} for range {}", keyspace, table, tokenRange); - try (Refs refs = getSSTableReaderRefs(repairType, keyspace, table, tokenRange)) - { - SizeEstimate estimate = getSizesForRangeOfSSTables(repairType, keyspace, table, tokenRange, refs); - logger.debug("Generated size estimate {}", estimate); - sizeEstimates.add(estimate); - } - return sizeEstimates; - } - - @VisibleForTesting - static SizeEstimate getSizesForRangeOfSSTables(AutoRepairConfig.RepairType repairType, String keyspace, String table, Range tokenRange, Refs refs) - { - List> singletonRange = Collections.singletonList(tokenRange); - ICardinality cardinality = new HyperLogLogPlus(13, 25); - long approxBytesInRange = 0L; - long totalBytes = 0L; - - for (SSTableReader reader : refs) - { - try - { - if (reader.openReason == SSTableReader.OpenReason.EARLY) - continue; - CompactionMetadata metadata = (CompactionMetadata) reader.descriptor.getMetadataSerializer().deserialize(reader.descriptor, MetadataType.COMPACTION); - if (metadata != null) - cardinality = cardinality.merge(metadata.cardinalityEstimator); - - // use onDiskLength, which is the actual size of the SSTable data file. - long sstableSize = reader.onDiskLength(); - totalBytes += sstableSize; - - // get the on disk size for the token range, note for compressed data this includes the full - // chunks the start and end ranges are found in. - long approximateRangeBytesInSSTable = reader.onDiskSizeForPartitionPositions(reader.getPositionsForRanges(singletonRange)); - approxBytesInRange += Math.min(approximateRangeBytesInSSTable, sstableSize); - } - catch (IOException | CardinalityMergeException e) - { - logger.error("Error calculating size estimate for {}.{} for range {} on {}", keyspace, table, tokenRange, reader, e); - } - } - - long partitions = 0L; - if (totalBytes > 0) - { - // use the ratio from size to estimate the partitions in the range as well - double ratio = approxBytesInRange / (double) totalBytes; - partitions = (long) Math.max(1, Math.ceil(cardinality.cardinality() * ratio)); - } - return new SizeEstimate(repairType, keyspace, table, tokenRange, partitions, approxBytesInRange, totalBytes); - } - @VisibleForTesting static Refs getSSTableReaderRefs(AutoRepairConfig.RepairType repairType, String keyspaceName, String tableName, Range tokenRange) { @@ -749,66 +665,15 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter return Collections.unmodifiableMap(parameters); } - /** - * Represents a size estimate by both bytes and partition count for a given keyspace and table for a token range. - */ - @VisibleForTesting - protected static class SizeEstimate - { - public final AutoRepairConfig.RepairType repairType; - public final String keyspace; - public final String table; - public final Range tokenRange; - public final long partitions; - public final long sizeInRange; - public final long totalSize; - /** - * Size to consider in the repair. For incremental repair, we want to consider the total size - * of the estimate as we have to factor in anticompacting the entire SSTable. - * For full repair, just use the size containing the range. - */ - public final long sizeForRepair; - - public SizeEstimate(AutoRepairConfig.RepairType repairType, - String keyspace, String table, Range tokenRange, - long partitions, long sizeInRange, long totalSize) - { - this.repairType = repairType; - this.keyspace = keyspace; - this.table = table; - this.tokenRange = tokenRange; - this.partitions = partitions; - this.sizeInRange = sizeInRange; - this.totalSize = totalSize; - - this.sizeForRepair = repairType == AutoRepairConfig.RepairType.INCREMENTAL ? totalSize : sizeInRange; - } - - @Override - public String toString() - { - return "SizeEstimate{" + - "repairType=" + repairType + - ", keyspace='" + keyspace + '\'' + - ", table='" + table + '\'' + - ", tokenRange=" + tokenRange + - ", partitions=" + partitions + - ", sizeInRange=" + sizeInRange + - ", totalSize=" + totalSize + - ", sizeForRepair=" + sizeForRepair + - '}'; - } - } - /** * Implementation of RepairAssignment that also assigns an estimation of bytes involved * in the repair. */ @VisibleForTesting - protected static class SizedRepairAssignment extends RepairAssignment { + protected static class SizedRepairAssignment extends RepairAssignment + { final String description; - final long estimatedBytes; public SizedRepairAssignment(Range tokenRange, String keyspaceName, List tableNames) { @@ -819,9 +684,8 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter String description, long estimatedBytes) { - super(tokenRange, keyspaceName, tableNames); + super(tokenRange, keyspaceName, tableNames, estimatedBytes); this.description = description; - this.estimatedBytes = estimatedBytes; } /** @@ -833,7 +697,8 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter } /** - * Estimated bytes involved in the assignment. Typically Derived from {@link SizeEstimate#sizeForRepair}. + * Estimated bytes involved in the assignment. Typically Derived from {@link AutoRepairUtils.SizeEstimate#sizeForRepair}. + * * @return estimated bytes involved in the assignment. */ public long getEstimatedBytes() diff --git a/test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerStatsHelper.java b/test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerStatsHelper.java new file mode 100644 index 0000000000..7e7089348e --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerStatsHelper.java @@ -0,0 +1,234 @@ +/* + * 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.distributed.test.repair; + +import java.io.IOException; +import java.text.ParseException; +import java.text.SimpleDateFormat; +import java.util.Arrays; +import java.util.List; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.TimeUnit; + +import com.google.common.collect.ImmutableMap; +import org.junit.Assert; + +import org.apache.cassandra.Util; +import org.apache.cassandra.auth.AuthKeyspace; +import org.apache.cassandra.config.CassandraRelevantProperties; +import org.apache.cassandra.config.DurationSpec; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.lifecycle.SSTableSet; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.TokenSupplier; +import org.apache.cassandra.distributed.test.TestBaseImpl; +import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.metrics.AutoRepairMetrics; +import org.apache.cassandra.metrics.AutoRepairMetricsManager; +import org.apache.cassandra.repair.autorepair.AutoRepair; +import org.apache.cassandra.repair.autorepair.AutoRepairConfig; +import org.apache.cassandra.schema.SchemaConstants; +import org.apache.cassandra.schema.SystemDistributedKeyspace; +import org.apache.cassandra.service.AutoRepairService; + +import static org.apache.cassandra.schema.SchemaConstants.DISTRIBUTED_KEYSPACE_NAME; +import static org.hamcrest.Matchers.greaterThanOrEqualTo; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +/** + * Helper class to test {@code totalBytesToRepair}, {@code bytesAlreadyRepaired}, {@code totalKeyspaceRepairPlansToRepair}, + * and {@code keyspaceRepairPlansAlreadyRepaired} + * for {@link org.apache.cassandra.repair.autorepair.AutoRepairState} scheduler + */ +public class AutoRepairSchedulerStatsHelper extends TestBaseImpl +{ + private static Cluster cluster; + static SimpleDateFormat sdf; + private static final String KEYSPACE1 = "ks1"; + private static final String KEYSPACE2 = "ks2"; + private static final String TABLE1 = "tbl1"; + private static final String TABLE2 = "tbl2"; + + public static void init(int numTokens) throws IOException + { + // Define the expected date format pattern + String pattern = "EEE MMM dd HH:mm:ss z yyyy"; + // Create SimpleDateFormat object with the given pattern + sdf = new SimpleDateFormat(pattern); + sdf.setLenient(false); + CassandraRelevantProperties.SYSTEM_DISTRIBUTED_DEFAULT_RF.setInt(1); + cluster = Cluster.build(1) + .withTokenCount(numTokens) + .withTokenSupplier(TokenSupplier.evenlyDistributedTokens(1, numTokens)) + .withConfig(config -> config + .set("num_tokens", numTokens) + .set("auto_repair", + ImmutableMap.of( + "repair_type_overrides", + ImmutableMap.of(AutoRepairConfig.RepairType.FULL.getConfigName(), + ImmutableMap.of( + "initial_scheduler_delay", "5s", + "enabled", "true", + "parallel_repair_count", "1", + // Allow parallel replica repair to allow replicas + // to execute full repair at same time. + "allow_parallel_replica_repair", "true", + // Set min_repair_interval to a higher number to + // run only one round of AutoRepair + "min_repair_interval", "48h")))) + .set("auto_repair.enabled", "true") + .set("auto_repair.global_settings.repair_retry_backoff", "5s") + .set("auto_repair.repair_task_min_duration", "0s") + .set("auto_repair.repair_check_interval", "5s")) + .start(); + + cluster.schemaChange("CREATE KEYSPACE IF NOT EXISTS " + KEYSPACE1 + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};"); + cluster.schemaChange("CREATE KEYSPACE IF NOT EXISTS " + KEYSPACE2 + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};"); + // disable the compression to calculate an accurate expected repair bytes because with compression enabled, + // we only get estimated bytes, which hinders the ability to do actual vs. expected checks in the test case + cluster.schemaChange(String.format("CREATE TABLE %s.%s (pk int, ck int, v int, PRIMARY KEY (pk, ck)) WITH compression = { 'enabled' : false }", KEYSPACE1, TABLE1)); + cluster.schemaChange(String.format("CREATE TABLE %s.%s (pk int, ck int, v int, PRIMARY KEY (pk, ck)) WITH compression = { 'enabled' : false }", KEYSPACE1, TABLE2)); + cluster.schemaChange(String.format("CREATE TABLE %s.%s (pk int, ck int, v int, PRIMARY KEY (pk, ck)) WITH compression = { 'enabled' : false }", KEYSPACE2, TABLE1)); + cluster.schemaChange(String.format("CREATE TABLE %s.%s (pk int, ck int, v int, PRIMARY KEY (pk, ck)) WITH compression = { 'enabled' : false }", KEYSPACE2, TABLE2)); + } + + public static void tearDown() + { + cluster.close(); + } + + public static void testSchedulerStats() throws ParseException + { + // ensure there was no history of previous repair runs through the scheduler + Object[][] rows = cluster.coordinator(1).execute(String.format("SELECT repair_type, host_id, repair_start_ts, repair_finish_ts, repair_turn FROM %s.%s", DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY), ConsistencyLevel.QUORUM); + assertEquals(0, rows.length); + + // disabling AutoRepair for system_distributed and system_auth tables to avoid + // interfering with the repaired bytes/plans calculation + disableAutoRepair(SystemDistributedKeyspace.NAME, SystemDistributedKeyspace.TABLE_NAMES); + disableAutoRepair(SchemaConstants.AUTH_KEYSPACE_NAME, AuthKeyspace.TABLE_NAMES); + + insertData(); + + cluster.get(1).runOnInstance(() -> { + try + { + AutoRepairService.setup(); + AutoRepair.instance.setup(); + } + catch (Exception e) + { + throw new RuntimeException(e); + } + }); + + cluster.forEach(i -> i.runOnInstance(() -> { + AutoRepair.SLEEP_IF_REPAIR_FINISHES_QUICKLY = new DurationSpec.IntSecondsBound("2s"); + + AutoRepairMetrics fullMetrics = AutoRepairMetricsManager.getMetrics(AutoRepairConfig.RepairType.FULL); + // Since the AutoRepair sleeps up to SLEEP_IF_REPAIR_FINISHES_QUICKLY if the repair finishes quickly, + // so the "nodeRepairTimeInSec" metric should at least be greater than or equal to + // SLEEP_IF_REPAIR_FINISHES_QUICKLY + Util.spinAssert("AutoRepair has not yet completed one FULL repair cycle", + greaterThanOrEqualTo(2L), + () -> fullMetrics.nodeRepairTimeInSec.getValue().longValue(), + 2, + TimeUnit.MINUTES); + + long expectedRepairBytes = calculateExpectedBytes(Arrays.asList(KEYSPACE1, KEYSPACE2)); + assertEquals(fullMetrics.totalKeyspaceRepairPlansToRepair.getValue(), fullMetrics.keyspaceRepairPlansAlreadyRepaired.getValue()); + // AutoRepair creates a repair plan per keyspace; + // Since there are two separate keyspaces, KEYSPACE1 and KEYSPACE2, the total expected plans should be "2" + assertEquals(2, fullMetrics.totalKeyspaceRepairPlansToRepair.getValue().intValue()); + assertEquals(fullMetrics.totalBytesToRepair.getValue().longValue(), fullMetrics.bytesAlreadyRepaired.getValue().longValue()); + assertEquals(expectedRepairBytes, fullMetrics.bytesAlreadyRepaired.getValue().longValue()); + })); + validate(AutoRepairConfig.RepairType.FULL.toString()); + } + + private static long calculateExpectedBytes(List keyspaces) + { + long totalBytes = 0; + for (String keyspace : keyspaces) + { + for (String table : Arrays.asList(TABLE1, TABLE2)) + { + ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(keyspace, table); + assertNotNull(cfs); + Iterable sstables = cfs.getTracker().getView().select(SSTableSet.CANONICAL); + for (SSTableReader sstable : sstables) + { + totalBytes += sstable.onDiskLength(); + } + } + } + return totalBytes; + } + + private static void insertData() + { + for (int i = 0; i < 100; i++) + { + cluster.coordinator(1).execute(String.format("INSERT INTO %s.%s (pk, ck, v) VALUES (?,?,?)", KEYSPACE1, TABLE1), + ConsistencyLevel.ONE, i, i, i); + cluster.coordinator(1).execute(String.format("INSERT INTO %s.%s (pk, ck, v) VALUES (?,?,?)", KEYSPACE1, TABLE2), + ConsistencyLevel.ONE, i, i, i); + cluster.coordinator(1).execute(String.format("INSERT INTO %s.%s (pk, ck, v) VALUES (?,?,?)", KEYSPACE2, TABLE1), + ConsistencyLevel.ONE, i, i, i); + cluster.coordinator(1).execute(String.format("INSERT INTO %s.%s (pk, ck, v) VALUES (?,?,?)", KEYSPACE2, TABLE2), + ConsistencyLevel.ONE, i, i, i); + } + cluster.get(1).nodetool("flush", KEYSPACE1, TABLE1); + cluster.get(1).nodetool("flush", KEYSPACE1, TABLE2); + cluster.get(1).nodetool("flush", KEYSPACE2, TABLE1); + cluster.get(1).nodetool("flush", KEYSPACE2, TABLE2); + } + + private static void disableAutoRepair(String keyspaceName, Set distributedSystemTables) + { + for (String tableName : distributedSystemTables) + { + cluster.coordinator(1).execute(String.format("ALTER TABLE %s.%s WITH auto_repair = {'full_enabled': 'false'}", keyspaceName, tableName), + ConsistencyLevel.ONE); + } + } + + private static void validate(String repairType) throws ParseException + { + Object[][] rows = cluster.coordinator(1).execute(String.format("SELECT repair_type, host_id, repair_start_ts, repair_finish_ts, repair_turn FROM %s.%s where repair_type='%s'", DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY, repairType), ConsistencyLevel.QUORUM); + assertEquals(1, rows.length); + for (int node = 0; node < rows.length; node++) + { + Object[] row = rows[node]; + // repair_type + Assert.assertEquals(repairType, row[0].toString()); + // host_id + Assert.assertNotNull(UUID.fromString(row[1].toString())); + // ensure there is a legit repair_start_ts and repair_finish_ts + sdf.parse(row[2].toString()); + sdf.parse(row[3].toString()); + // the reason why the repair was scheduled + Assert.assertNotNull(row[4]); + Assert.assertEquals("MY_TURN", row[4].toString()); + } + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerStatsNoVNodesTest.java b/test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerStatsNoVNodesTest.java new file mode 100644 index 0000000000..96376095b9 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerStatsNoVNodesTest.java @@ -0,0 +1,54 @@ +/* + * 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.distributed.test.repair; + +import java.io.IOException; +import java.text.ParseException; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.distributed.test.TestBaseImpl; + +/** + * Test {@code totalBytesToRepair}, {@code bytesAlreadyRepaired}, {@code totalKeyspaceRepairPlansToRepair}, + * and {@code keyspaceRepairPlansAlreadyRepaired} + * for {@link org.apache.cassandra.repair.autorepair.AutoRepairState} scheduler without v-nodes + */ +public class AutoRepairSchedulerStatsNoVNodesTest extends TestBaseImpl +{ + @BeforeClass + public static void init() throws IOException + { + AutoRepairSchedulerStatsHelper.init(1); + } + + @AfterClass + public static void tearDown() + { + AutoRepairSchedulerStatsHelper.tearDown(); + } + + @Test + public void testSchedulerStats() throws ParseException + { + AutoRepairSchedulerStatsHelper.testSchedulerStats(); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerStatsVNodesTest.java b/test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerStatsVNodesTest.java new file mode 100644 index 0000000000..8040a30328 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerStatsVNodesTest.java @@ -0,0 +1,54 @@ +/* + * 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.distributed.test.repair; + +import java.io.IOException; +import java.text.ParseException; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.distributed.test.TestBaseImpl; + +/** + * Test {@code totalBytesToRepair}, {@code bytesAlreadyRepaired}, {@code totalKeyspaceRepairPlansToRepair}, + * and {@code keyspaceRepairPlansAlreadyRepaired} + * for {@link org.apache.cassandra.repair.autorepair.AutoRepairState} scheduler with v-nodes + */ +public class AutoRepairSchedulerStatsVNodesTest extends TestBaseImpl +{ + @BeforeClass + public static void init() throws IOException + { + AutoRepairSchedulerStatsHelper.init(16); + } + + @AfterClass + public static void tearDown() + { + AutoRepairSchedulerStatsHelper.tearDown(); + } + + @Test + public void testSchedulerStats() throws ParseException + { + AutoRepairSchedulerStatsHelper.testSchedulerStats(); + } +} diff --git a/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairStateFactoryTest.java b/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairStateFactoryTest.java index 97e80364ee..91c2c39ee4 100644 --- a/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairStateFactoryTest.java +++ b/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairStateFactoryTest.java @@ -34,15 +34,15 @@ public class AutoRepairStateFactoryTest @Test public void testGetRepairState() { - AutoRepairState state = RepairType.getAutoRepairState(RepairType.FULL); + AutoRepairState state = RepairType.getAutoRepairState(RepairType.FULL, new AutoRepairConfig()); assertTrue(state instanceof FullRepairState); - state = RepairType.getAutoRepairState(RepairType.INCREMENTAL); + state = RepairType.getAutoRepairState(RepairType.INCREMENTAL, new AutoRepairConfig()); assertTrue(state instanceof IncrementalRepairState); - state = RepairType.getAutoRepairState(RepairType.PREVIEW_REPAIRED); + state = RepairType.getAutoRepairState(RepairType.PREVIEW_REPAIRED, new AutoRepairConfig()); assertTrue(state instanceof PreviewRepairedState); } @@ -54,7 +54,7 @@ public class AutoRepairStateFactoryTest { try { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); assertNotNull(state); } catch (IllegalArgumentException e) { diff --git a/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairStateTest.java b/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairStateTest.java index 422ebdff51..240a2db46f 100644 --- a/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairStateTest.java +++ b/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairStateTest.java @@ -34,8 +34,6 @@ import org.apache.cassandra.cql3.CQLTester; import org.apache.cassandra.repair.autorepair.AutoRepairConfig.RepairType; import org.apache.cassandra.repair.autorepair.AutoRepairUtils.AutoRepairHistory; import org.apache.cassandra.service.AutoRepairService; -import org.apache.cassandra.utils.progress.ProgressEvent; -import org.mockito.Mock; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; @@ -54,9 +52,6 @@ public class AutoRepairStateTest extends CQLTester @Parameterized.Parameter public RepairType repairType; - @Mock - ProgressEvent progressEvent; - @Parameterized.Parameters public static Collection repairTypes() { @@ -74,7 +69,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testGetRepairRunnable() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); AutoRepairService.setup(); Runnable runnable = state.getRepairRunnable(KEYSPACE, ImmutableList.of(testTable), ImmutableSet.of(), false); @@ -85,7 +80,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testGetLastRepairTime() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.lastRepairTimeInMs = 1; assertEquals(1, state.getLastRepairTime()); @@ -94,7 +89,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testSetTotalTablesConsideredForRepair() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.setTotalTablesConsideredForRepair(1); @@ -104,7 +99,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testGetTotalTablesConsideredForRepair() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.totalTablesConsideredForRepair = 1; assertEquals(1, state.getTotalTablesConsideredForRepair()); @@ -113,7 +108,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testSetLastRepairTimeInMs() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.setLastRepairTime(1); @@ -123,7 +118,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testGetClusterRepairTimeInSec() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.clusterRepairTimeInSec = 1; assertEquals(1, state.getClusterRepairTimeInSec()); @@ -132,7 +127,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testGetNodeRepairTimeInSec() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.nodeRepairTimeInSec = 1; assertEquals(1, state.getNodeRepairTimeInSec()); @@ -141,7 +136,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testSetRepairInProgress() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.setRepairInProgress(true); @@ -151,7 +146,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testIsRepairInProgress() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.repairInProgress = true; assertTrue(state.isRepairInProgress()); @@ -160,7 +155,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testSetSkippedTokenRangesCount() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.setSkippedTokenRangesCount(1); @@ -170,7 +165,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testGetSkippedTokenRangesCount() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.skippedTokenRangesCount = 1; assertEquals(1, state.getSkippedTokenRangesCount()); @@ -179,7 +174,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testGetLongestUnrepairedSecNull() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.longestUnrepairedNode = null; try @@ -195,7 +190,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testGetLongestUnrepairedSec() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.longestUnrepairedNode = new AutoRepairHistory(UUID.randomUUID(), "", 0, 1000, null, 0, false); AutoRepairState.timeFunc = () -> 2000L; @@ -213,7 +208,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testSetTotalMVTablesConsideredForRepair() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.setTotalMVTablesConsideredForRepair(1); @@ -223,7 +218,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testGetTotalMVTablesConsideredForRepair() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.totalMVTablesConsideredForRepair = 1; assertEquals(1, state.getTotalMVTablesConsideredForRepair()); @@ -232,7 +227,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testSetNodeRepairTimeInSec() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.setNodeRepairTimeInSec(1); @@ -242,7 +237,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testSetClusterRepairTimeInSec() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.setClusterRepairTimeInSec(1); @@ -252,7 +247,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testSetRepairKeyspaceCount() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.setRepairKeyspaceCount(1); @@ -262,7 +257,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testGetRepairKeyspaceCount() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.repairKeyspaceCount = 1; assertEquals(1, state.getRepairKeyspaceCount()); @@ -271,7 +266,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testSetLongestUnrepairedNode() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); AutoRepairHistory history = new AutoRepairHistory(UUID.randomUUID(), "", 0, 0, null, 0, false); state.setLongestUnrepairedNode(history); @@ -282,7 +277,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testSetSucceededTokenRangesCount() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.setSucceededTokenRangesCount(1); @@ -292,7 +287,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testGetSucceededTokenRangesCount() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.succeededTokenRangesCount = 1; assertEquals(1, state.getSucceededTokenRangesCount()); @@ -301,7 +296,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testSetFailedTokenRangesCount() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.setFailedTokenRangesCount(1); @@ -311,7 +306,7 @@ public class AutoRepairStateTest extends CQLTester @Test public void testGetFailedTokenRangesCount() { - AutoRepairState state = RepairType.getAutoRepairState(repairType); + AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig()); state.failedTokenRangesCount = 1; assertEquals(1, state.getFailedTokenRangesCount()); diff --git a/test/unit/org/apache/cassandra/repair/autorepair/FixedSplitTokenRangeSplitterHelper.java b/test/unit/org/apache/cassandra/repair/autorepair/FixedSplitTokenRangeSplitterHelper.java index dac4a167d5..ccb8d63a40 100644 --- a/test/unit/org/apache/cassandra/repair/autorepair/FixedSplitTokenRangeSplitterHelper.java +++ b/test/unit/org/apache/cassandra/repair/autorepair/FixedSplitTokenRangeSplitterHelper.java @@ -22,8 +22,10 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.HashMap; import java.util.Iterator; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.TreeSet; @@ -39,11 +41,14 @@ import org.apache.cassandra.service.StorageService; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.compatibility.TokenRingUtils; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.Pair; import static org.apache.cassandra.config.CassandraRelevantProperties.SYSTEM_DISTRIBUTED_DEFAULT_RF; import static org.apache.cassandra.cql3.CQLTester.Fuzzed.setupSeed; import static org.apache.cassandra.cql3.CQLTester.Fuzzed.updateConfigs; +import static org.apache.cassandra.repair.autorepair.AutoRepairUtils.split; import static org.apache.cassandra.repair.autorepair.FixedSplitTokenRangeSplitter.DEFAULT_NUMBER_OF_SUBRANGES; +import static org.apache.cassandra.repair.autorepair.FixedSplitTokenRangeSplitter.NUMBER_OF_SUBRANGES; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; @@ -58,6 +63,8 @@ public class FixedSplitTokenRangeSplitterHelper private static final String TABLE2 = "tbl2"; private static final String TABLE3 = "tbl3"; public static final String KEYSPACE = "ks"; + public static final List tables = Arrays.asList(TABLE1, TABLE2, TABLE3); + public static final Map, AutoRepairUtils.SizeEstimate>> ksTablesEstimatedBytes = new HashMap<>(); public static void setupClass(int numTokens) throws Exception { @@ -74,35 +81,59 @@ public class FixedSplitTokenRangeSplitterHelper SYSTEM_DISTRIBUTED_DEFAULT_RF.setInt(1); QueryProcessor.executeInternal(String.format("CREATE KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}", FixedSplitTokenRangeSplitterHelper.KEYSPACE)); + + Pair>, Integer> tokensAndWrappedAroundCount = getTokenRangesAndTotalWrapAroundCount(); + int totalToken = numTokens + tokensAndWrappedAroundCount.right(); + long perTokenSizeTable1 = 512L / totalToken; + long perTokenSizeTable2 = 1024L / totalToken; + long perTokenSizeTable3 = 2048L / totalToken; + for (Range tokenRange : tokensAndWrappedAroundCount.left) + { + ksTablesEstimatedBytes.put(AutoRepairUtils.getKeyspaceTableName(KEYSPACE, TABLE1), new HashMap<>() + {{ + put(tokenRange, new AutoRepairUtils.SizeEstimate(AutoRepairConfig.RepairType.FULL, "", "", tokenRange, 0, perTokenSizeTable1, perTokenSizeTable1)); + }}); + ksTablesEstimatedBytes.put(AutoRepairUtils.getKeyspaceTableName(KEYSPACE, TABLE2), new HashMap<>() + {{ + put(tokenRange, new AutoRepairUtils.SizeEstimate(AutoRepairConfig.RepairType.FULL, "", "", tokenRange, 0, perTokenSizeTable2, perTokenSizeTable2)); + }}); + ksTablesEstimatedBytes.put(AutoRepairUtils.getKeyspaceTableName(KEYSPACE, TABLE3), new HashMap<>() + {{ + put(tokenRange, new AutoRepairUtils.SizeEstimate(AutoRepairConfig.RepairType.FULL, "", "", tokenRange, 0, perTokenSizeTable3, perTokenSizeTable3)); + }}); + } } public static void testTokenRangesSplitByTable(int numTokens, int numberOfSubRanges, AutoRepairConfig.RepairType repairType) { - int numberOfSplits = calcSplits(numTokens, numberOfSubRanges); - AutoRepairService.instance.getAutoRepairConfig().setRepairByKeyspace(repairType, false); - Collection> tokens = TokenRingUtils.getPrimaryRangesForEndpoint(KEYSPACE, FBUtilities.getBroadcastAddressAndPort()); + AutoRepairConfig config = AutoRepairService.instance.getAutoRepairConfig(); + config.setRepairByKeyspace(repairType, false); + Pair>, Integer> tokensAndWrappedAroundCount = getTokenRangesAndTotalWrapAroundCount(); + Collection> tokens = tokensAndWrappedAroundCount.left(); + // For the test case, the tokens are allocated dynamically, so we do not know which token-ranges wrap around. + // As a result, we need to adjust the token count on a need basis. + numTokens += tokensAndWrappedAroundCount.right(); assertEquals(numTokens, tokens.size()); - List tables = Arrays.asList(TABLE1, TABLE2, TABLE3); List> expectedToken = new ArrayList<>(); + int numberOfSplits = Math.max(1, numberOfSubRanges / tokens.size()); for (int i = 0; i < tables.size(); i++) { - for (Range range : tokens) + for (Range token : tokens) { - expectedToken.addAll(AutoRepairUtils.split(range, numberOfSplits)); + expectedToken.addAll(split(token, numberOfSplits)); } } - List plan = PrioritizedRepairPlan.buildSingleKeyspacePlan(repairType, KEYSPACE, TABLE1, TABLE2, TABLE3); - - Iterator keyspaceAssignments = new FixedSplitTokenRangeSplitter(repairType, Collections.singletonMap(FixedSplitTokenRangeSplitter.NUMBER_OF_SUBRANGES, Integer.toString(numberOfSubRanges))) - .getRepairAssignments(true, plan); + Iterator keyspaceAssignments = + new FixedSplitTokenRangeSplitter(repairType, Collections.singletonMap(NUMBER_OF_SUBRANGES, Integer.toString(numberOfSubRanges))) + .getRepairAssignments(config.getRepairPrimaryTokenRangeOnly(repairType), getPlan(repairType)); // should be only 1 entry for the keyspace. assertTrue(keyspaceAssignments.hasNext()); - KeyspaceRepairAssignments keyspace = keyspaceAssignments.next(); + KeyspaceRepairAssignments keyspaceRepairAssignment = keyspaceAssignments.next(); assertFalse(keyspaceAssignments.hasNext()); - List assignments = keyspace.getRepairAssignments(); + List assignments = keyspaceRepairAssignment.getRepairAssignments(); assertEquals(numTokens * numberOfSplits * tables.size(), assignments.size()); assertEquals(expectedToken.size(), assignments.size()); @@ -113,9 +144,12 @@ public class FixedSplitTokenRangeSplitterHelper List> expectedTokensForATable = new ArrayList<>(); for (int j = 0; j < assignmentsPerTable; j++) { - assertEquals(Collections.singletonList(tables.get(i)), assignments.get(i * assignmentsPerTable + j).getTableNames()); - assignmentForATable.add(assignments.get(i * assignmentsPerTable + j)); - expectedTokensForATable.add(expectedToken.get(i * assignmentsPerTable + j)); + long expectedBytes = ksTablesEstimatedBytes.get(AutoRepairUtils.getKeyspaceTableName(KEYSPACE, tables.get(i))).values().stream().mapToLong(sizeEstimate -> sizeEstimate.sizeForRepair).sum() / numberOfSplits; + int theTableAssignmentIdx = i * assignmentsPerTable + j; + assertEquals(expectedBytes, assignments.get(theTableAssignmentIdx).estimatedBytes); + assertEquals(Collections.singletonList(tables.get(i)), assignments.get(theTableAssignmentIdx).getTableNames()); + assignmentForATable.add(assignments.get(theTableAssignmentIdx)); + expectedTokensForATable.add(expectedToken.get(theTableAssignmentIdx)); } compare(numTokens, numberOfSplits, expectedTokensForATable, assignmentForATable); } @@ -123,20 +157,24 @@ public class FixedSplitTokenRangeSplitterHelper public static void testTokenRangesSplitByKeyspace(int numTokens, int numberOfSubRanges, AutoRepairConfig.RepairType repairType) { - int numberOfSplits = calcSplits(numTokens, numberOfSubRanges); - AutoRepairService.instance.getAutoRepairConfig().setRepairByKeyspace(repairType, true); - Collection> tokens = TokenRingUtils.getPrimaryRangesForEndpoint(KEYSPACE, FBUtilities.getBroadcastAddressAndPort()); + AutoRepairConfig config = AutoRepairService.instance.getAutoRepairConfig(); + config.setRepairByKeyspace(repairType, true); + Pair>, Integer> tokensAndWrappedRanges = getTokenRangesAndTotalWrapAroundCount(); + Collection> tokens = tokensAndWrappedRanges.left(); + // For the test case, the tokens are allocated dynamically, so we do not know which token-ranges wrap around. + // As a result, we need to adjust the token count on a need basis. + numTokens += tokensAndWrappedRanges.right(); assertEquals(numTokens, tokens.size()); + int numberOfSplits = Math.max(1, numberOfSubRanges / tokens.size()); List> expectedToken = new ArrayList<>(); for (Range range : tokens) { expectedToken.addAll(AutoRepairUtils.split(range, numberOfSplits)); } - List plan = PrioritizedRepairPlan.buildSingleKeyspacePlan(repairType, KEYSPACE, TABLE1, TABLE2, TABLE3); - - Iterator keyspaceAssignments = new FixedSplitTokenRangeSplitter(repairType, Collections.singletonMap(FixedSplitTokenRangeSplitter.NUMBER_OF_SUBRANGES, Integer.toString(numberOfSubRanges))) - .getRepairAssignments(true, plan); + Iterator keyspaceAssignments = + new FixedSplitTokenRangeSplitter(repairType, Collections.singletonMap(NUMBER_OF_SUBRANGES, Integer.toString(numberOfSubRanges))) + .getRepairAssignments(config.getRepairPrimaryTokenRangeOnly(repairType), getPlan(repairType)); // should be only 1 entry for the keyspace. assertTrue(keyspaceAssignments.hasNext()); @@ -150,35 +188,20 @@ public class FixedSplitTokenRangeSplitterHelper assertEquals(expectedToken.size(), assignments.size()); compare(numTokens, numberOfSplits, expectedToken, assignments); + + for (int i = 0; i < assignments.size(); i++) + { + assertEquals(assignments.get(i).estimatedBytes, + ksTablesEstimatedBytes.values().stream() + .flatMap(tableMap -> tableMap.values().stream()) + .mapToLong(sizeEstimate -> sizeEstimate.sizeForRepair) + .sum() / numberOfSplits); + } } public static void testTokenRangesWithDefaultSplit(int numTokens, AutoRepairConfig.RepairType repairType) { - int numberOfSplits = calcSplits(numTokens, DEFAULT_NUMBER_OF_SUBRANGES); - Collection> tokens = TokenRingUtils.getPrimaryRangesForEndpoint(KEYSPACE, FBUtilities.getBroadcastAddressAndPort()); - assertEquals(numTokens, tokens.size()); - List> expectedToken = new ArrayList<>(); - for (Range range : tokens) - { - expectedToken.addAll(AutoRepairUtils.split(range, numberOfSplits)); - } - - List plan = PrioritizedRepairPlan.buildSingleKeyspacePlan(repairType, KEYSPACE, TABLE1); - - Iterator keyspaceAssignments = new FixedSplitTokenRangeSplitter(repairType, Collections.emptyMap()).getRepairAssignments(true, plan); - - // should be only 1 entry for the keyspace. - assertTrue(keyspaceAssignments.hasNext()); - KeyspaceRepairAssignments keyspace = keyspaceAssignments.next(); - assertFalse(keyspaceAssignments.hasNext()); - - List assignments = keyspace.getRepairAssignments(); - assertNotNull(assignments); - - // should be 3 entries for the table which covers each token range. - assertEquals(numTokens * numberOfSplits, assignments.size()); - - compare(numTokens, numberOfSplits, expectedToken, assignments); + testTokenRangesSplitByKeyspace(numTokens, DEFAULT_NUMBER_OF_SUBRANGES, repairType); } private static void compare(int numTokens, int numberOfSplits, List> expectedToken, List assignments) @@ -194,8 +217,34 @@ public class FixedSplitTokenRangeSplitterHelper assertEquals(a, b); } - private static int calcSplits(int numTokens, int subRange) + private static Pair>, Integer> getTokenRangesAndTotalWrapAroundCount() { - return Math.max(1, subRange / numTokens); + int wrappedRanges = 0; + Collection> ranges = TokenRingUtils.getPrimaryRangesForEndpoint(KEYSPACE, FBUtilities.getBroadcastAddressAndPort()); + Collection> tokens = new ArrayList<>(); + for (Range wrappedRange : ranges) + { + if (wrappedRange.isWrapAround()) + { + wrappedRanges++; + } + tokens.addAll(wrappedRange.unwrap()); + } + return Pair.create(tokens, wrappedRanges); + } + + private static List getPlan(AutoRepairConfig.RepairType repairType) + { + AutoRepairConfig config = AutoRepairService.instance.getAutoRepairConfig(); + List plan = PrioritizedRepairPlan.build(new HashMap<>() + {{ + put(KEYSPACE, tables); + }}, repairType, (l) -> { + }, + config.getRepairPrimaryTokenRangeOnly(repairType)); + assertEquals(1, plan.size()); + assertEquals(1, plan.get(0).getKeyspaceRepairPlans().size()); + plan.get(0).getKeyspaceRepairPlans().get(0).ksTablesEstimatedBytes = ksTablesEstimatedBytes; + return plan; } } diff --git a/test/unit/org/apache/cassandra/repair/autorepair/PrioritizedRepairPlanTest.java b/test/unit/org/apache/cassandra/repair/autorepair/PrioritizedRepairPlanTest.java index 38f9c85388..be9e93aeca 100644 --- a/test/unit/org/apache/cassandra/repair/autorepair/PrioritizedRepairPlanTest.java +++ b/test/unit/org/apache/cassandra/repair/autorepair/PrioritizedRepairPlanTest.java @@ -18,14 +18,18 @@ package org.apache.cassandra.repair.autorepair; +import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; import com.google.common.collect.Lists; +import org.junit.BeforeClass; import org.junit.Test; import org.apache.cassandra.cql3.CQLTester; +import org.apache.cassandra.service.AutoRepairService; +import org.apache.cassandra.service.StorageService; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -35,6 +39,13 @@ import static org.junit.Assert.assertTrue; */ public class PrioritizedRepairPlanTest extends CQLTester { + + @BeforeClass + public static void setup() + { + StorageService.instance.doAutoRepairSetup(); + } + @Test public void testBuildWithDifferentPriorities() { @@ -43,7 +54,8 @@ public class PrioritizedRepairPlanTest extends CQLTester String table2 = createTable("CREATE TABLE %s (k INT PRIMARY KEY, v INT) WITH auto_repair = {'full_enabled': 'true', 'priority': '3'}"); String table3 = createTable("CREATE TABLE %s (k INT PRIMARY KEY, v INT) WITH auto_repair = {'full_enabled': 'true', 'priority': '1'}"); - List prioritizedRepairPlans = PrioritizedRepairPlan.buildSingleKeyspacePlan(AutoRepairConfig.RepairType.FULL, KEYSPACE, table1, table2, table3); + List prioritizedRepairPlans = PrioritizedRepairPlan.build(new HashMap<>(){{put(KEYSPACE, Arrays.asList(table1, table2, table3));}}, AutoRepairConfig.RepairType.FULL, (l) -> {}, + AutoRepairService.instance.getAutoRepairConfig().getRepairPrimaryTokenRangeOnly(AutoRepairConfig.RepairType.FULL)); assertEquals(3, prioritizedRepairPlans.size()); // Verify the order is by descending priority and matches the expected tables @@ -66,7 +78,8 @@ public class PrioritizedRepairPlanTest extends CQLTester String table3 = createTable("CREATE TABLE %s (k INT PRIMARY KEY, v INT) WITH auto_repair = {'full_enabled': 'true', 'priority': '2'}"); // Expect only 1 plan since all tables share the same priority - List prioritizedRepairPlans = PrioritizedRepairPlan.buildSingleKeyspacePlan(AutoRepairConfig.RepairType.FULL, KEYSPACE, table1, table2, table3); + List prioritizedRepairPlans = PrioritizedRepairPlan.build(new HashMap<>(){{put(KEYSPACE, Arrays.asList(table1, table2, table3));}}, AutoRepairConfig.RepairType.FULL, (l) -> {}, + AutoRepairService.instance.getAutoRepairConfig().getRepairPrimaryTokenRangeOnly(AutoRepairConfig.RepairType.FULL)); assertEquals(1, prioritizedRepairPlans.size()); // Verify all tables present in the plan @@ -101,7 +114,7 @@ public class PrioritizedRepairPlanTest extends CQLTester keyspaceToTableMap.put(ks2, Lists.newArrayList(table6, table7)); // Expect 4 plans - List prioritizedRepairPlans = PrioritizedRepairPlan.build(keyspaceToTableMap, AutoRepairConfig.RepairType.FULL, java.util.Collections::sort); + List prioritizedRepairPlans = PrioritizedRepairPlan.build(keyspaceToTableMap, AutoRepairConfig.RepairType.FULL, java.util.Collections::sort, true); assertEquals(4, prioritizedRepairPlans.size()); // Verify the order is by descending priority and matches the expected tables @@ -143,7 +156,8 @@ public class PrioritizedRepairPlanTest extends CQLTester public void testBuildWithEmptyTableList() { // Test with an empty table list (should remain empty) - List prioritizedRepairPlans = PrioritizedRepairPlan.buildSingleKeyspacePlan(AutoRepairConfig.RepairType.FULL, KEYSPACE); + List prioritizedRepairPlans = PrioritizedRepairPlan.build(new HashMap<>(){{put(KEYSPACE, Arrays.asList());}}, AutoRepairConfig.RepairType.FULL, (l) -> {}, + AutoRepairService.instance.getAutoRepairConfig().getRepairPrimaryTokenRangeOnly(AutoRepairConfig.RepairType.FULL)); assertTrue(prioritizedRepairPlans.isEmpty()); } @@ -154,7 +168,8 @@ public class PrioritizedRepairPlanTest extends CQLTester String table1 = createTable("CREATE TABLE %s (k INT PRIMARY KEY, v INT) WITH auto_repair = {'full_enabled': 'true', 'priority': '5'}"); // Expect only 1 plans - List prioritizedRepairPlans = PrioritizedRepairPlan.buildSingleKeyspacePlan(AutoRepairConfig.RepairType.FULL, KEYSPACE, table1); + List prioritizedRepairPlans = PrioritizedRepairPlan.build(new HashMap<>(){{put(KEYSPACE, Arrays.asList(table1));}}, AutoRepairConfig.RepairType.FULL, (l) -> {}, + AutoRepairService.instance.getAutoRepairConfig().getRepairPrimaryTokenRangeOnly(AutoRepairConfig.RepairType.FULL)); assertEquals(1, prioritizedRepairPlans.size()); // Verify the order is by descending priority and matches the expected tables diff --git a/test/unit/org/apache/cassandra/repair/autorepair/RepairTokenRangeSplitterTest.java b/test/unit/org/apache/cassandra/repair/autorepair/RepairTokenRangeSplitterTest.java index 79fef533f1..6351f4fd97 100644 --- a/test/unit/org/apache/cassandra/repair/autorepair/RepairTokenRangeSplitterTest.java +++ b/test/unit/org/apache/cassandra/repair/autorepair/RepairTokenRangeSplitterTest.java @@ -22,6 +22,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; import java.util.List; @@ -39,6 +40,7 @@ import org.apache.cassandra.config.DataStorageSpec.LongMebibytesBound; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.cql3.CQLTester; import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.lifecycle.SSTableSet; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.io.sstable.format.SSTableReader; @@ -46,11 +48,13 @@ import org.apache.cassandra.io.sstable.format.big.BigFormat; import org.apache.cassandra.io.sstable.format.bti.BtiFormat; import org.apache.cassandra.repair.autorepair.AutoRepairConfig.RepairType; import org.apache.cassandra.repair.autorepair.RepairTokenRangeSplitter.FilteredRepairAssignments; -import org.apache.cassandra.repair.autorepair.RepairTokenRangeSplitter.SizeEstimate; +import org.apache.cassandra.repair.autorepair.AutoRepairUtils.SizeEstimate; import org.apache.cassandra.repair.autorepair.RepairTokenRangeSplitter.SizedRepairAssignment; +import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.service.AutoRepairService; import org.apache.cassandra.utils.concurrent.Refs; +import static org.apache.cassandra.repair.autorepair.AutoRepairUtils.getKeyspaceTableName; import static org.apache.cassandra.repair.autorepair.RepairTokenRangeSplitter.MAX_BYTES_PER_SCHEDULE; import static org.apache.cassandra.repair.autorepair.RepairTokenRangeSplitter.BYTES_PER_ASSIGNMENT; import static org.apache.cassandra.repair.autorepair.RepairTokenRangeSplitter.MAX_TABLES_PER_ASSIGNMENT; @@ -111,7 +115,7 @@ public class RepairTokenRangeSplitterTest extends CQLTester try (Refs sstables = RepairTokenRangeSplitter.getSSTableReaderRefs(RepairType.FULL, KEYSPACE, tableName, FULL_RANGE)) { assertEquals(10, sstables.iterator().next().getEstimatedPartitionSize().count()); - SizeEstimate sizes = RepairTokenRangeSplitter.getSizesForRangeOfSSTables(RepairType.FULL, KEYSPACE, tableName, FULL_RANGE, sstables); + SizeEstimate sizes = AutoRepairUtils.getSizesForRangeOfSSTables(RepairType.FULL, KEYSPACE, tableName, FULL_RANGE, sstables); assertEquals(10, sizes.partitions); } } @@ -132,8 +136,8 @@ public class RepairTokenRangeSplitterTest extends CQLTester try (Refs sstables1 = RepairTokenRangeSplitter.getSSTableReaderRefs(RepairType.FULL, KEYSPACE, tableName, tokenRange1); Refs sstables2 = RepairTokenRangeSplitter.getSSTableReaderRefs(RepairType.FULL, KEYSPACE, tableName, tokenRange2)) { - SizeEstimate sizes1 = RepairTokenRangeSplitter.getSizesForRangeOfSSTables(RepairType.FULL, KEYSPACE, tableName, tokenRange1, sstables1); - SizeEstimate sizes2 = RepairTokenRangeSplitter.getSizesForRangeOfSSTables(RepairType.FULL, KEYSPACE, tableName, tokenRange2, sstables2); + SizeEstimate sizes1 = AutoRepairUtils.getSizesForRangeOfSSTables(RepairType.FULL, KEYSPACE, tableName, tokenRange1, sstables1); + SizeEstimate sizes2 = AutoRepairUtils.getSizesForRangeOfSSTables(RepairType.FULL, KEYSPACE, tableName, tokenRange2, sstables2); // +-5% because including entire compression blocks covering token range, HLL merge and the applying of range size approx ratio causes estimation errors long allowableDelta = (long) (partitionCount * .05); @@ -143,68 +147,150 @@ public class RepairTokenRangeSplitterTest extends CQLTester } @Test - public void testGetRepairAssignmentsForTable_NoSSTables() + public void testGetRepairAssignmentsForTableNoSSTables() { // Should return 1 assignment if there are no SSTables - List assignments = repairRangeSplitter.getRepairAssignmentsForTable(CQLTester.KEYSPACE, tableName, FULL_RANGE); + List assignments = repairRangeSplitter.getRepairAssignmentsForTable(new KeyspaceRepairPlan(CQLTester.KEYSPACE, Collections.singletonList(tableName), AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, Collections.singletonList(tableName), Collections.singletonList(FULL_RANGE))), tableName, FULL_RANGE); assertEquals(1, assignments.size()); } @Test - public void testGetRepairAssignmentsForTable_Single() + public void testGetRepairAssignmentsForTableSingle() { insertAndFlushSingleTable(); - List assignments = repairRangeSplitter.getRepairAssignmentsForTable(CQLTester.KEYSPACE, tableName, FULL_RANGE); + List assignments = repairRangeSplitter.getRepairAssignmentsForTable(new KeyspaceRepairPlan(CQLTester.KEYSPACE, Collections.singletonList(tableName), AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, Collections.singletonList(tableName), Collections.singletonList(FULL_RANGE))), tableName, FULL_RANGE); assertEquals(1, assignments.size()); } @Test - public void testGetRepairAssignmentsForTable_BatchingTables() + public void testGetRepairAssignmentsForTableBatchingTablesCompressed() { repairRangeSplitter = new RepairTokenRangeSplitter(RepairType.FULL, Collections.singletonMap(MAX_TABLES_PER_ASSIGNMENT, "2")); - List tableNames = createAndInsertTables(3); - List assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, KEYSPACE, tableNames, FULL_RANGE); + List tableNames = createAndInsertTables(3, true); + List assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, new KeyspaceRepairPlan(CQLTester.KEYSPACE, tableNames, AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, tableNames, Collections.singletonList(FULL_RANGE))), FULL_RANGE); // We expect two assignments, one with table1 and table2 batched, and one with table3 assertEquals(2, assignments.size()); assertEquals(2, assignments.get(0).getTableNames().size()); assertEquals(1, assignments.get(1).getTableNames().size()); + assertEquals(new HashSet<>(Arrays.asList(tableNames.get(0), tableNames.get(1))), new HashSet<>(assignments.get(0).getTableNames())); + assertEquals(tableNames.get(2), assignments.get(1).getTableNames().get(0)); } @Test - public void testGetRepairAssignmentsForTable_BatchSize() + public void testGetRepairAssignmentsForTableBatchingTablesUncompressed() { repairRangeSplitter = new RepairTokenRangeSplitter(RepairType.FULL, Collections.singletonMap(MAX_TABLES_PER_ASSIGNMENT, "2")); - List tableNames = createAndInsertTables(2); - List assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, KEYSPACE, tableNames, FULL_RANGE); + List tableNames = createAndInsertTables(3, false); + List assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, new KeyspaceRepairPlan(CQLTester.KEYSPACE, tableNames, AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, tableNames, Collections.singletonList(FULL_RANGE))), FULL_RANGE); + + // We expect two assignments, one with table1 and table2 batched, and one with table3 + assertEquals(2, assignments.size()); + assertEquals(2, assignments.get(0).getTableNames().size()); + assertEquals(1, assignments.get(1).getTableNames().size()); + assertEquals(new HashSet<>(Arrays.asList(tableNames.get(0), tableNames.get(1))), new HashSet<>(assignments.get(0).getTableNames())); + assertEquals(tableNames.get(2), assignments.get(1).getTableNames().get(0)); + assertTrue(assignments.get(0).getEstimatedBytes() > 0); + assertTrue(assignments.get(1).getEstimatedBytes() > 0); + assertEquals(calculateSSTableSizeOnDisk(Arrays.asList(tableNames.get(0), tableNames.get(1))), assignments.get(0).getEstimatedBytes()); + assertEquals(calculateSSTableSizeOnDisk(Collections.singletonList(tableNames.get(2))), assignments.get(1).getEstimatedBytes()); + } + + @Test + public void testGetRepairAssignmentsForTableBatchSizeCompressed() + { + repairRangeSplitter = new RepairTokenRangeSplitter(RepairType.FULL, Collections.singletonMap(MAX_TABLES_PER_ASSIGNMENT, "2")); + + List tableNames = createAndInsertTables(2, true); + List assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, new KeyspaceRepairPlan(CQLTester.KEYSPACE, tableNames, AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, tableNames, Collections.singletonList(FULL_RANGE))), FULL_RANGE); // We expect one assignment, with two tables batched assertEquals(1, assignments.size()); assertEquals(2, assignments.get(0).getTableNames().size()); + assertEquals(new HashSet<>(Arrays.asList(tableNames.get(0), tableNames.get(1))), new HashSet<>(assignments.get(0).getTableNames())); } @Test - public void testGetRepairAssignmentsForTable_NoBatching() + public void testGetRepairAssignmentsForTableBatchSizeUnCompressed() + { + repairRangeSplitter = new RepairTokenRangeSplitter(RepairType.FULL, Collections.singletonMap(MAX_TABLES_PER_ASSIGNMENT, "2")); + + List tableNames = createAndInsertTables(2, false); + List assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, + new KeyspaceRepairPlan(CQLTester.KEYSPACE, tableNames, AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, tableNames, Collections.singletonList(FULL_RANGE))), FULL_RANGE); + + // We expect one assignment, with two tables batched + assertEquals(1, assignments.size()); + assertEquals(2, assignments.get(0).getTableNames().size()); + assertEquals(new HashSet<>(Arrays.asList(tableNames.get(0), tableNames.get(1))), new HashSet<>(assignments.get(0).getTableNames())); + assertTrue(assignments.get(0).getEstimatedBytes() > 0); + assertEquals(calculateSSTableSizeOnDisk(Arrays.asList(tableNames.get(0), tableNames.get(1))), assignments.get(0).getEstimatedBytes()); + } + + @Test + public void testGetRepairAssignmentsForTableNoBatchingCompressed() { repairRangeSplitter = new RepairTokenRangeSplitter(RepairType.FULL, Collections.singletonMap(MAX_TABLES_PER_ASSIGNMENT, "1")); - List tableNames = createAndInsertTables(3); - List assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, KEYSPACE, tableNames, FULL_RANGE); + List tableNames = createAndInsertTables(3, true); + List assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, new KeyspaceRepairPlan(CQLTester.KEYSPACE, tableNames, AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, tableNames, Collections.singletonList(FULL_RANGE))), FULL_RANGE); assertEquals(3, assignments.size()); + assertEquals(Collections.singletonList(tableNames.get(0)), assignments.get(0).getTableNames()); + assertEquals(Collections.singletonList(tableNames.get(1)), assignments.get(1).getTableNames()); + assertEquals(Collections.singletonList(tableNames.get(2)), assignments.get(2).getTableNames()); } @Test - public void testGetRepairAssignmentsForTable_AllBatched() + public void testGetRepairAssignmentsForTableNoBatchingUncompressed() + { + repairRangeSplitter = new RepairTokenRangeSplitter(RepairType.FULL, Collections.singletonMap(MAX_TABLES_PER_ASSIGNMENT, "1")); + + List tableNames = createAndInsertTables(3, false); + List assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, new KeyspaceRepairPlan(CQLTester.KEYSPACE, tableNames, AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, tableNames, Collections.singletonList(FULL_RANGE))), FULL_RANGE); + + assertEquals(3, assignments.size()); + assertEquals(Collections.singletonList(tableNames.get(0)), assignments.get(0).getTableNames()); + assertEquals(Collections.singletonList(tableNames.get(1)), assignments.get(1).getTableNames()); + assertEquals(Collections.singletonList(tableNames.get(2)), assignments.get(2).getTableNames()); + assertTrue(assignments.get(0).getEstimatedBytes() > 0); + assertTrue(assignments.get(1).getEstimatedBytes() > 0); + assertTrue(assignments.get(2).getEstimatedBytes() > 0); + assertEquals(calculateSSTableSizeOnDisk(Collections.singletonList(tableNames.get(0))), assignments.get(0).getEstimatedBytes()); + assertEquals(calculateSSTableSizeOnDisk(Collections.singletonList(tableNames.get(1))), assignments.get(1).getEstimatedBytes()); + assertEquals(calculateSSTableSizeOnDisk(Collections.singletonList(tableNames.get(2))), assignments.get(2).getEstimatedBytes()); + } + + @Test + public void testGetRepairAssignmentsForTableAllBatchedCompressed() { repairRangeSplitter = new RepairTokenRangeSplitter(RepairType.FULL, Collections.singletonMap(MAX_TABLES_PER_ASSIGNMENT, "100")); - List tableNames = createAndInsertTables(5); - List assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, KEYSPACE, tableNames, FULL_RANGE); + List tableNames = createAndInsertTables(5, true); + List assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, new KeyspaceRepairPlan(CQLTester.KEYSPACE, tableNames, AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, tableNames, Collections.singletonList(FULL_RANGE))), FULL_RANGE); assertEquals(1, assignments.size()); + assertEquals(5, assignments.get(0).getTableNames().size()); + assertEquals(new HashSet<>(tableNames), + new HashSet<>(assignments.get(0).getTableNames())); + } + + @Test + public void testGetRepairAssignmentsForTableAllBatchedUncompressed() + { + repairRangeSplitter = new RepairTokenRangeSplitter(RepairType.FULL, Collections.singletonMap(MAX_TABLES_PER_ASSIGNMENT, "100")); + + List tableNames = createAndInsertTables(5, false); + List assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, new KeyspaceRepairPlan(CQLTester.KEYSPACE, tableNames, AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, tableNames, Collections.singletonList(FULL_RANGE))), FULL_RANGE); + + assertEquals(1, assignments.size()); + assertEquals(5, assignments.get(0).getTableNames().size()); + assertEquals(new HashSet<>(tableNames), + new HashSet<>(assignments.get(0).getTableNames())); + assertTrue(assignments.get(0).getEstimatedBytes() > 0); + assertEquals(calculateSSTableSizeOnDisk(tableNames), assignments.get(0).getEstimatedBytes()); } @Test(expected = IllegalStateException.class) @@ -312,7 +398,7 @@ public class RepairTokenRangeSplitterTest extends CQLTester // Given a size estimate of 1024GiB, we should expect 21 splits (50GiB*21 = 1050GiB < 1024GiB) SizeEstimate sizeEstimate = sizeEstimateByBytes(new LongMebibytesBound("1024GiB")); - List assignments = repairRangeSplitter.getRepairAssignments(Collections.singletonList(sizeEstimate)); + List assignments = repairRangeSplitter.getRepairAssignments(sizeEstimate); // Should be 21 assignments, each being ~48.76 GiB assertEquals(21, assignments.size()); @@ -335,7 +421,24 @@ public class RepairTokenRangeSplitterTest extends CQLTester { AutoRepairService.instance.getAutoRepairConfig().setRepairByKeyspace(RepairType.FULL, true); - final KeyspaceRepairPlan repairPlan = new KeyspaceRepairPlan("system_auth", new ArrayList<>(AuthKeyspace.TABLE_NAMES)); + Map, AutoRepairUtils.SizeEstimate>> ksTablesEstimatedBytes = new HashMap<>(); + List> tokenRanges = AutoRepairUtils.getTokenRanges(true, SchemaConstants.AUTH_KEYSPACE_NAME); + long tableSizeInBytes = 100L; + long tableSizeInBytesPerTokenRange = tableSizeInBytes / tokenRanges.size(); + for (String tableName : AuthKeyspace.TABLE_NAMES) + { + String ksTableName = getKeyspaceTableName(SchemaConstants.AUTH_KEYSPACE_NAME, tableName); + ksTablesEstimatedBytes.putIfAbsent(ksTableName, new HashMap<>()); + Map, AutoRepairUtils.SizeEstimate> rangeSizeEstimateMap = ksTablesEstimatedBytes.get(ksTableName); + + for (Range tokenRange : tokenRanges) + { + rangeSizeEstimateMap.put(tokenRange, new AutoRepairUtils.SizeEstimate(AutoRepairConfig.RepairType.FULL, SchemaConstants.AUTH_KEYSPACE_NAME, tableName, tokenRange, 0, tableSizeInBytesPerTokenRange, tableSizeInBytesPerTokenRange)); + } + } + final KeyspaceRepairPlan repairPlan = new KeyspaceRepairPlan(SchemaConstants.AUTH_KEYSPACE_NAME, new ArrayList<>(AuthKeyspace.TABLE_NAMES), ksTablesEstimatedBytes); + assertEquals(tableSizeInBytes * AuthKeyspace.TABLE_NAMES.size(), repairPlan.getEstimatedBytes()); + final PrioritizedRepairPlan prioritizedRepairPlan = new PrioritizedRepairPlan(0, List.of(repairPlan)); Iterator keyspaceAssignments = repairRangeSplitter.getRepairAssignments(true, List.of(prioritizedRepairPlan)); @@ -363,7 +466,25 @@ public class RepairTokenRangeSplitterTest extends CQLTester { AutoRepairService.instance.getAutoRepairConfig().setRepairByKeyspace(RepairType.FULL, false); - final KeyspaceRepairPlan repairPlan = new KeyspaceRepairPlan("system_auth", new ArrayList<>(AuthKeyspace.TABLE_NAMES)); + Map, AutoRepairUtils.SizeEstimate>> ksTablesEstimatedBytes = new HashMap<>(); + List> tokenRanges = AutoRepairUtils.getTokenRanges(true, SchemaConstants.AUTH_KEYSPACE_NAME); + long tableSizeInBytes = 100L; + long tableSizeInBytesPerTokenRange = tableSizeInBytes / tokenRanges.size(); + for (String tableName : AuthKeyspace.TABLE_NAMES) + { + String ksTableName = getKeyspaceTableName(SchemaConstants.AUTH_KEYSPACE_NAME, tableName); + ksTablesEstimatedBytes.putIfAbsent(ksTableName, new HashMap<>()); + Map, AutoRepairUtils.SizeEstimate> rangeSizeEstimateMap = ksTablesEstimatedBytes.get(ksTableName); + + for (Range tokenRange : tokenRanges) + { + rangeSizeEstimateMap.put(tokenRange, new AutoRepairUtils.SizeEstimate(AutoRepairConfig.RepairType.FULL, SchemaConstants.AUTH_KEYSPACE_NAME, tableName, tokenRange, 0, tableSizeInBytesPerTokenRange, tableSizeInBytesPerTokenRange)); + } + } + + final KeyspaceRepairPlan repairPlan = new KeyspaceRepairPlan(SchemaConstants.AUTH_KEYSPACE_NAME, new ArrayList<>(AuthKeyspace.TABLE_NAMES), ksTablesEstimatedBytes); + assertEquals(tableSizeInBytes * AuthKeyspace.TABLE_NAMES.size(), repairPlan.getEstimatedBytes()); + final PrioritizedRepairPlan prioritizedRepairPlan = new PrioritizedRepairPlan(0, List.of(repairPlan)); Iterator keyspaceAssignments = repairRangeSplitter.getRepairAssignments(true, List.of(prioritizedRepairPlan)); @@ -436,18 +557,43 @@ public class RepairTokenRangeSplitterTest extends CQLTester flush(); } - private List createAndInsertTables(int count) + private List createAndInsertTables(int count, boolean enableCompression) { List tableNames = new ArrayList<>(); for (int i = 0; i < count; i++) { - String tableName = createTable("CREATE TABLE %s (k INT PRIMARY KEY, v INT)"); + String tableName; + if (enableCompression) + { + tableName = createTable("CREATE TABLE %s (k INT PRIMARY KEY, v INT)"); + } + else + { + tableName = createTable("CREATE TABLE %s (k INT PRIMARY KEY, v INT) WITH compression = { 'enabled' : false }"); + } tableNames.add(tableName); insertAndFlushTable(tableName); } return tableNames; } + private long calculateSSTableSizeOnDisk(List tableNames) + { + long totalSSTableBytes = 0; + for (int i = 0; i < tableNames.size(); i++) + { + String tableName = tableNames.get(i); + ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, tableName); + assertNotNull(cfs); + Iterable sstables = cfs.getTracker().getView().select(SSTableSet.CANONICAL); + for (SSTableReader sstable : sstables) + { + totalSSTableBytes += sstable.onDiskLength(); + } + } + return totalSSTableBytes; + } + private void insertAndFlushTable(String tableName) { insertAndFlushTable(tableName, 1);