C20048 multi version repair guardrail trunk

conf/cassandra.yaml

@@ -1969,6 +1969,13 @@ report_unconfirmed_repaired_data_mismatches: false
 # Materialized views are considered experimental and are not recommended for production use.
 materialized_views_enabled: false
 
+# Specify whether Materialized View mutations are replayed through the write path on streaming, e.g. repair.
+# When enabled, Materialized View data streamed to the destination node will be written into commit log first. When setting to false,
+# the streamed Materialized View data is written into SSTables just the same as normal streaming. The default is true.
+# If this is set to false, streaming will be considerably faster however it's possible that, in extreme situations
+# (losing > quorum # nodes in a replica set), you may have data in your SSTables that never makes it to the Materialized View.
+# materialized_views_on_repair_enabled: true
+
 # Enables SASI index creation on this node.
 # SASI indexes are considered experimental and are not recommended for production use.
 sasi_indexes_enabled: false

doc/modules/cassandra/pages/managing/operating/metrics.adoc

@@ -246,12 +246,19 @@ during validation.
 |PartitionsValidated |Histogram |Histogram over the number of partitions
 read during validation.
 
-|BytesAnticompacted |Counter |How many bytes we anticompacted.
+|BytesAnticompacted |Meter |How many bytes we anticompacted.
 
-|BytesMutatedAnticompaction |Counter |How many bytes we avoided
+|BytesMutatedAnticompaction |Meter |How many bytes we avoided
 anticompacting because the sstable was fully contained in the repaired
 range.
 
+|BytesPreviewed |Meter |Estimated number of bytes that were scanned for local replica during preview repair
+
+|BytesPreviewedDesynchronized |Meter |Number of desynchronized bytes that were detected among all replicas during preview repair
+
+|TokenRangesPreviewedDesynchronized |Meter |Number of token ranges among all replicas where desynchronization was found
+during preview repair. These ranges would need to be streamed during subsequent repair.
+
 |MutatedAnticompactionGauge |Gauge<Double> |Ratio of bytes mutated vs
 total bytes repaired.
 |===
@@ -1078,6 +1085,60 @@ partitions processed per logged batch
 partitions processed per unlogged batch
 |===
 
+== Automated Repair Metrics
+
+Metrics specifc to automated repair.
+
+Reported name format:
+
+*Metric Name*::
+`org.apache.cassandra.metrics.AutoRepair.<MetricName>`
+*JMX MBean*::
+`org.apache.cassandra.metrics:type=AutoRepair name=<MetricName> repairType=<RepairType>`
+
+[cols=",,",options="header",]
+|===
+|Name |Type |Description
+|RepairsInProgress |Gauge<Integer> |Repair is in progress
+on the node
+
+|NodeRepairTimeInSec |Gauge<Integer> |Time taken to repair
+the node in seconds
+
+|ClusterRepairTimeInSec |Gauge<Integer> |Time taken to repair
+the entire Cassandra cluster in seconds
+
+|LongestUnrepairedSec |Gauge<Integer> |Time since the last repair
+ran on the node in seconds
+
+|SucceededTokenRangesCount |Gauge<Integer> |Number of token ranges successfully repaired on the node
+
+|FailedTokenRangesCount |Gauge<Integer> |Number of token ranges failed to repair on the node
+
+|SkippedTokenRangesCount |Gauge<Integer> |Number of token ranges skipped
+on the node
+
+|SkippedTablesCount |Gauge<Integer> |Number of tables skipped
+on the node
+
+|TotalMVTablesConsideredForRepair |Gauge<Integer> |Number of materialized
+views considered on the node
+
+|TotalDisabledRepairTables |Gauge<Integer> |Number of tables on which
+the automated repair has been disabled on the node
+
+|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
+
+|RepairTurnMyTurnForceRepair |Counter |Represents the node's turn to repair
+due to force repair set in the automated repair
+
+
+|===
+
+
 == JVM Metrics
 
 JVM metrics such as memory and garbage collection statistics can either

pylib/cqlshlib/test/test_cqlsh_output.py

@@ -698,7 +698,8 @@ def test_describe_columnfamily_output(self):
                 AND memtable_flush_period_in_ms = 0
                 AND min_index_interval = 128
                 AND read_repair = 'BLOCKING'
-                AND speculative_retry = '99p';""" % quote_name(get_keyspace()))
+                AND speculative_retry = '99p'
+                AND auto_repair = {'full_enabled': 'true', 'incremental_enabled': 'true', 'preview_repaired_enabled': 'true', 'priority': '0'};""" % quote_name(get_keyspace()))
 
         with cqlsh_testrun(tty=True, env=self.default_env) as c:
             for cmdword in ('describe table', 'desc columnfamily'):

src/java/org/apache/cassandra/config/Config.java

@@ -32,6 +32,8 @@
 import com.google.common.base.Joiner;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Sets;
+
+import org.apache.cassandra.repair.autorepair.AutoRepairConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -353,6 +355,10 @@ public MemtableOptions()
     // The number of executors to use for building secondary indexes
     public volatile int concurrent_index_builders = 2;
 
+    // at least 20% of disk must be unused to run incremental repair
+    // if you want to disable this feature (the recommendation is not to, but if you want to disable it for whatever reason) then set the ratio to 0.0
+    public volatile double incremental_repair_disk_headroom_reject_ratio = 0.2;
+
     /**
      * @deprecated retry support removed on CASSANDRA-10992
      */
@@ -417,6 +423,7 @@ public static class SSTableConfig
     // When false, it behaves the same as normal streaming.
     public volatile boolean cdc_on_repair_enabled = true;
     public String cdc_raw_directory;
+    public boolean mixed_version_repairs_enabled = true;
     @Replaces(oldName = "cdc_total_space_in_mb", converter = Converters.MEBIBYTES_DATA_STORAGE_INT, deprecated = true)
     public DataStorageSpec.IntMebibytesBound cdc_total_space = new DataStorageSpec.IntMebibytesBound("0MiB");
     @Replaces(oldName = "cdc_free_space_check_interval_ms", converter = Converters.MILLIS_DURATION_INT, deprecated = true)
@@ -595,6 +602,10 @@ public static class SSTableConfig
     @Replaces(oldName = "enable_materialized_views", converter = Converters.IDENTITY, deprecated = true)
     public boolean materialized_views_enabled = false;
 
+    // When true, materialized views data in SSTable go through commit logs during internodes streaming, e.g. repair
+    // When false, it behaves the same as normal streaming.
+    public volatile boolean materialized_views_on_repair_enabled = true;
+
     @Replaces(oldName = "enable_transient_replication", converter = Converters.IDENTITY, deprecated = true)
     public boolean transient_replication_enabled = false;
 
@@ -947,6 +958,7 @@ public static void setClientMode(boolean clientMode)
 
     public volatile boolean password_validator_reconfiguration_enabled = true;
     public volatile CustomGuardrailConfig password_validator = new CustomGuardrailConfig();
+    public volatile AutoRepairConfig auto_repair = new AutoRepairConfig();
 
     /**
      * The variants of paxos implementation and semantics supported by Cassandra.

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -107,6 +107,7 @@
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.locator.Replica;
 import org.apache.cassandra.locator.SeedProvider;
+import org.apache.cassandra.repair.autorepair.AutoRepairConfig;
 import org.apache.cassandra.security.AbstractCryptoProvider;
 import org.apache.cassandra.security.EncryptionContext;
 import org.apache.cassandra.security.JREProvider;
@@ -4206,6 +4207,16 @@ public static void setMaterializedViewsEnabled(boolean enableMaterializedViews)
         conf.materialized_views_enabled = enableMaterializedViews;
     }
 
+    public static boolean isMaterializedViewsOnRepairEnabled()
+    {
+        return conf.materialized_views_on_repair_enabled;
+    }
+
+    public static void setMaterializedViewsOnRepairEnabled(boolean val)
+    {
+        conf.materialized_views_on_repair_enabled = val;
+    }
+
     public static boolean getSASIIndexesEnabled()
     {
         return conf.sasi_indexes_enabled;
@@ -5349,4 +5360,25 @@ public static boolean isPasswordValidatorReconfigurationEnabled()
     {
         return conf.password_validator_reconfiguration_enabled;
     }
+
+    public static AutoRepairConfig getAutoRepairConfig()
+    {
+        return conf.auto_repair;
+    }
+
+    public static double getIncrementalRepairDiskHeadroomRejectRatio()
+    {
+        return conf.incremental_repair_disk_headroom_reject_ratio;
+    }
+
+    public static void setIncrementalRepairDiskHeadroomRejectRatio(double value)
+    {
+        conf.incremental_repair_disk_headroom_reject_ratio = value;
+    }
+
+    @VisibleForTesting
+    public static void setPartitioner(String name)
+    {
+        partitioner = FBUtilities.newPartitioner(name);
+    }
 }

src/java/org/apache/cassandra/config/DurationSpec.java

@@ -17,6 +17,7 @@
  */
 package org.apache.cassandra.config;
 
+import java.io.Serializable;
 import java.util.Arrays;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
@@ -41,7 +42,7 @@
  * users the opportunity to be able to provide config with a unit of their choice in cassandra.yaml as per the available
  * options. (CASSANDRA-15234)
  */
-public abstract class DurationSpec
+public abstract class DurationSpec implements Serializable
 {
     /**
      * The Regexp used to parse the duration provided as String.

src/java/org/apache/cassandra/config/GuardrailsOptions.java

@@ -205,6 +205,22 @@ public int getMaterializedViewsPerTableWarnThreshold()
         return config.materialized_views_per_table_warn_threshold;
     }
 
+    @Override
+    public boolean getMixedVersionRepairsEnabled()
+    {
+        return config.mixed_version_repairs_enabled;
+    }
+
+    @Override
+    public boolean setMixedVersionRepairsEnabled(boolean enabled)
+    {
+        updatePropertyWithLogging("mixed_version_repairs_enabled",
+                                  enabled,
+                                  () -> config.mixed_version_repairs_enabled,
+                                  x -> config.mixed_version_repairs_enabled = x);
+        return enabled;
+    }
+
     @Override
     public int getPartitionKeysInSelectWarnThreshold()
     {

src/java/org/apache/cassandra/config/ParameterizedClass.java

@@ -19,6 +19,7 @@
 
 import java.lang.reflect.Constructor;
 import java.util.Collections;
+import java.io.Serializable;
 import java.util.List;
 import java.util.Map;
 
@@ -30,7 +31,7 @@
 import static org.apache.cassandra.utils.Shared.Scope.SIMULATION;
 
 @Shared(scope = SIMULATION)
-public class ParameterizedClass
+public class ParameterizedClass implements Serializable
 {
     public static final String CLASS_NAME = "class_name";
     public static final String PARAMETERS = "parameters";

src/java/org/apache/cassandra/cql3/UntypedResultSet.java

@@ -449,6 +449,13 @@ public long getLong(String column)
             return LongType.instance.compose(data.get(column));
         }
 
+        // this function will return the default value if the row doesn't have that column or the column data is null
+        // This function is used to avoid the nullpointerexception
+        public long getLong(String column, long ifNull) {
+            ByteBuffer bytes = data.get(column);
+            return bytes == null ? ifNull : LongType.instance.compose(bytes);
+        }
+
         public <T> Set<T> getSet(String column, AbstractType<T> type)
         {
             ByteBuffer raw = data.get(column);

src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java

@@ -81,7 +81,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
     public static final String CUSTOM_EXPRESSIONS_NOT_ALLOWED =
         "Custom index expressions cannot be used in WHERE clauses for UPDATE or DELETE statements";
 
-    private static final ColumnIdentifier CAS_RESULT_COLUMN = new ColumnIdentifier("[applied]", false);
+    public static final ColumnIdentifier CAS_RESULT_COLUMN = new ColumnIdentifier("[applied]", false);
 
     protected final StatementType type;
 

src/java/org/apache/cassandra/cql3/statements/schema/TableAttributes.java

@@ -25,6 +25,7 @@
 
 import org.apache.cassandra.cql3.statements.PropertyDefinitions;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.schema.AutoRepairParams;
 import org.apache.cassandra.schema.CachingParams;
 import org.apache.cassandra.schema.CompactionParams;
 import org.apache.cassandra.schema.CompressionParams;
@@ -151,6 +152,9 @@ private TableParams build(TableParams.Builder builder)
         if (hasOption(READ_REPAIR))
             builder.readRepair(ReadRepairStrategy.fromString(getString(READ_REPAIR)));
 
+        if (hasOption(Option.AUTO_REPAIR))
+            builder.automatedRepair(AutoRepairParams.fromMap(getMap(Option.AUTO_REPAIR)));
+
         return builder.build();
     }
 

src/java/org/apache/cassandra/db/compaction/CompactionManager.java

@@ -888,7 +888,7 @@ private static void mutateFullyContainedSSTables(ColumnFamilyStore cfs,
 
         Set<SSTableReader> fullyContainedSSTables = findSSTablesToAnticompact(sstableIterator, normalizedRanges, sessionID);
 
-        cfs.metric.bytesMutatedAnticompaction.inc(SSTableReader.getTotalBytes(fullyContainedSSTables));
+        cfs.metric.bytesMutatedAnticompaction.mark(SSTableReader.getTotalBytes(fullyContainedSSTables));
         cfs.getCompactionStrategyManager().mutateRepaired(fullyContainedSSTables, UNREPAIRED_SSTABLE, sessionID, isTransient);
         // since we're just re-writing the sstable metdata for the fully contained sstables, we don't want
         // them obsoleted when the anti-compaction is complete. So they're removed from the transaction here
@@ -1724,7 +1724,7 @@ private void doAntiCompaction(ColumnFamilyStore cfs,
         // repairedAt values for these, we still avoid anti-compacting already repaired sstables, as we currently don't
         // make use of any actual repairedAt value and splitting up sstables just for that is not worth it at this point.
         Set<SSTableReader> unrepairedSSTables = sstables.stream().filter((s) -> !s.isRepaired()).collect(Collectors.toSet());
-        cfs.metric.bytesAnticompacted.inc(SSTableReader.getTotalBytes(unrepairedSSTables));
+        cfs.metric.bytesAnticompacted.mark(SSTableReader.getTotalBytes(unrepairedSSTables));
         Collection<Collection<SSTableReader>> groupedSSTables = cfs.getCompactionStrategyManager().groupSSTablesForAntiCompaction(unrepairedSSTables);
 
         // iterate over sstables to check if the full / transient / unrepaired ranges intersect them.

src/java/org/apache/cassandra/db/guardrails/Guardrails.java

@@ -39,7 +39,12 @@
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.disk.usage.DiskUsageBroadcaster;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.membership.Directory;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.tcm.membership.NodeVersion;
 import org.apache.cassandra.utils.MBeanWrapper;
+import org.apache.cassandra.utils.btree.BTreeMap;
 
 import static java.lang.String.format;
 
@@ -289,6 +294,23 @@ public final class Guardrails implements GuardrailsMBean
                    state -> CONFIG_PROVIDER.getOrCreate(state).getAllowFilteringEnabled(),
                    "Querying with ALLOW FILTERING");
 
+    /**
+     * Guardrail disabling repairs when there are mixed versions
+     */
+    public static final EnableFlag mixedRepairsEnabled =
+    new EnableFlag("mixed_version_repairs",
+                    "Mixed mode repairs and streaming adds many unknowns and additional performance impacts during upgrades.",
+                    state -> {
+                        if (CONFIG_PROVIDER.getOrCreate(state).getMixedVersionRepairsEnabled())
+                            return true;
+                        Directory directory = ClusterMetadata.current().directory;
+                        Set<NodeId> nodes = directory.states.keySet();
+                        BTreeMap<NodeId, NodeVersion> versions = directory.versions;
+                        NodeVersion version = versions.get(nodes.iterator().next());
+                        return nodes.stream().allMatch(node -> versions.get(node).equals(version));
+                    },
+                    "Running repairs during mixed Cassandra versions");
+
     /**
      * Guardrail disabling setting SimpleStrategy via keyspace creation or alteration
      */
@@ -594,6 +616,18 @@ public void setKeyspacesThreshold(int warn, int fail)
         DEFAULT_CONFIG.setKeyspacesThreshold(warn, fail);
     }
 
+    @Override
+    public boolean getMixedVersionRepairsEnabled()
+    {
+        return DEFAULT_CONFIG.getMixedVersionRepairsEnabled();
+    }
+
+    @Override
+    public void setMixedVersionRepairsEnabled(boolean enabled)
+    {
+        DEFAULT_CONFIG.setMixedVersionRepairsEnabled(enabled);
+    }
+
     @Override
     public int getTablesWarnThreshold()
     {

src/java/org/apache/cassandra/db/guardrails/GuardrailsConfig.java

@@ -106,6 +106,12 @@ public interface GuardrailsConfig
      */
     int getPartitionKeysInSelectFailThreshold();
 
+    /**
+     * @return if repairs during mixed versions are allowed
+     */
+    boolean getMixedVersionRepairsEnabled();
+    boolean setMixedVersionRepairsEnabled(boolean enabled);
+
     /**
      * @return The threshold to fail when creating more materialized views per table than threshold.
      */