CASSANDRA-21024: Add configuration to disk usage guardrails to stop writes across all replicas of a keyspace when any node replicating that keyspace exceeds the disk usage failure threshold.

This commit adds a new configuration, data_disk_usage_keyspace_wide_protection_enabled, which ensures that if any node which replicates a given keyspace is full, all writes to that keyspace are blocked.

patch by Isaac Reath; reviewed by Stefan Miklosovic, Paulo Motta for CASSANDRA-20124

Closes #4547

Author: isaacreath

CHANGES.txt

@@ -1,4 +1,5 @@
 5.1
+ * Add configuration to disk usage guardrails to stop writes across all replicas of a keyspace when any node replicating that keyspace exceeds the disk usage failure threshold. (CASSANDRA-21024)
  * BETWEEN where token(Y) > token(Z) returns wrong answer (CASSANDRA-20154)
  * Optimize memtable flush logic (CASSANDRA-21083)
  * No need to evict already prepared statements, as it creates a race condition between multiple threads (CASSANDRA-17401)

conf/cassandra.yaml

@@ -2517,6 +2517,10 @@ drop_compact_storage_enabled: false
 # Min unit: B
 # data_disk_usage_max_disk_size:
 #
+# Configures the disk usage guardrails to block all writes to a keyspace if any node which replicates that keyspace
+# is full. By default, this is disabled.
+# data_disk_usage_keyspace_wide_protection_enabled: false
+#
 # Guardrail to warn or fail when the minimum replication factor is lesser than threshold.
 # This would also apply to system keyspaces.
 # Suggested value for use in production: 2 or higher

conf/cassandra_latest.yaml

@@ -2295,6 +2295,10 @@ drop_compact_storage_enabled: false
 # Min unit: B
 # data_disk_usage_max_disk_size:
 #
+# Configures the disk usage guardrails to block all writes to a keyspace if any node which replicates that keyspace
+# is full. By default, this is disabled.
+# data_disk_usage_keyspace_wide_protection_enabled: false
+#
 # Guardrail to warn or fail when the minimum replication factor is lesser than threshold.
 # This would also apply to system keyspaces.
 # Suggested value for use in production: 2 or higher

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

@@ -985,6 +985,7 @@ public static void setClientMode(boolean clientMode)
     public volatile int data_disk_usage_percentage_warn_threshold = -1;
     public volatile int data_disk_usage_percentage_fail_threshold = -1;
     public volatile DataStorageSpec.LongBytesBound data_disk_usage_max_disk_size = null;
+    public volatile boolean data_disk_usage_keyspace_wide_protection_enabled = false;
     public volatile int minimum_replication_factor_warn_threshold = -1;
     public volatile int minimum_replication_factor_fail_threshold = -1;
     public volatile int maximum_replication_factor_warn_threshold = -1;

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

@@ -989,6 +989,20 @@ public void setDataDiskUsagePercentageThreshold(int warn, int fail)
                                   x -> config.data_disk_usage_percentage_fail_threshold = x);
     }
 
+
+    public boolean getDataDiskUsageKeyspaceWideProtectionEnabled()
+    {
+        return config.data_disk_usage_keyspace_wide_protection_enabled;
+    }
+
+    public void setDataDiskUsageKeyspaceWideProtectionEnabled(boolean enabled)
+    {
+        updatePropertyWithLogging("data_disk_usage_keyspace_wide_protection_enabled",
+                                  enabled,
+                                  () -> config.data_disk_usage_keyspace_wide_protection_enabled,
+                                  x -> config.data_disk_usage_keyspace_wide_protection_enabled = x);
+    }
+
     @Override
     public DataStorageSpec.LongBytesBound getDataDiskUsageMaxDiskSize()
     {

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

@@ -103,6 +103,7 @@
 import org.apache.cassandra.exceptions.RequestExecutionException;
 import org.apache.cassandra.exceptions.RequestValidationException;
 import org.apache.cassandra.exceptions.UnauthorizedException;
+import org.apache.cassandra.locator.NetworkTopologyStrategy;
 import org.apache.cassandra.locator.Replica;
 import org.apache.cassandra.locator.ReplicaLayout;
 import org.apache.cassandra.metrics.ClientRequestSizeMetrics;
@@ -124,6 +125,7 @@
 import org.apache.cassandra.service.paxos.Ballot;
 import org.apache.cassandra.service.paxos.BallotGenerator;
 import org.apache.cassandra.service.paxos.Commit.Proposal;
+import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.transport.Dispatcher;
 import org.apache.cassandra.transport.messages.ResultMessage;
 import org.apache.cassandra.triggers.TriggerExecutor;
@@ -412,15 +414,36 @@ public void validate(ClientState state) throws InvalidRequestException
 
     public void validateDiskUsage(QueryOptions options, ClientState state)
     {
-        // reject writes if any replica exceeds disk usage failure limit or warn if it exceeds warn limit
-        if (Guardrails.replicaDiskUsage.enabled(state) && DiskUsageBroadcaster.instance.hasStuffedOrFullNode())
+
+        if (Guardrails.diskUsageKeyspaceWideProtection.enabled(state) &&
+            Guardrails.instance.getDataDiskUsageKeyspaceWideProtectionEnabled() &&
+            DiskUsageBroadcaster.instance.hasStuffedOrFullNode())
+        {
+            Keyspace keyspace = Keyspace.open(keyspace());
+            // If the keyspace is using NetworkTopologyStrategy then we can check each datacenter on which
+            // the keyspace is replicated.
+            if (keyspace.getMetadata().replicationStrategy instanceof NetworkTopologyStrategy)
+            {
+                for (String datacenter : ((NetworkTopologyStrategy) keyspace.getMetadata().replicationStrategy).getDatacenters())
+                {
+                    Guardrails.diskUsageKeyspaceWideProtection.guard(datacenter, state);
+                }
+            }
+            // Otherwise, if we are using SimpleStrategy then we have to check if any datacenter contains a full node.
+            else
+            {
+                for (String datacenter : ClusterMetadata.current().directory.knownDatacenters())
+                {
+                    Guardrails.diskUsageKeyspaceWideProtection.guard(datacenter, state);
+                }
+            }
+        }
+        else if (Guardrails.replicaDiskUsage.enabled(state) && DiskUsageBroadcaster.instance.hasStuffedOrFullNode())
         {
             Keyspace keyspace = Keyspace.open(keyspace());
-
             for (ByteBuffer key : buildPartitionKeyNames(options, state))
             {
                 Token token = metadata().partitioner.getToken(key);
-
                 for (Replica replica : ReplicaLayout.forTokenWriteLiveAndDown(keyspace, token).all())
                 {
                     Guardrails.replicaDiskUsage.guard(replica.endpoint(), state);

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

@@ -556,6 +556,19 @@ public final class Guardrails implements GuardrailsMBean
                      (isWarning, value) ->
                      isWarning ? "Replica disk usage exceeds warning threshold"
                                : "Write request failed because disk usage exceeds failure threshold");
+    /**
+     * Guardrail on the data disk usage of replicas across a datacenter which replicates a given keyspace.
+     * This is used at write time to verify the status of any node which might replicate a given keyspace.
+     */
+    public static final Predicates<String> diskUsageKeyspaceWideProtection =
+    new Predicates<>("disk_usage_keyspace_wide_protection",
+                     null,
+                     state -> DiskUsageBroadcaster.instance::isDatacenterStuffed,
+                     state -> DiskUsageBroadcaster.instance::isDatacenterFull,
+                     (isWarning, value) ->
+                     isWarning ? "Disk usage in keyspace datacenter exceeds warning threshold"
+                               : "Write request failed because disk usage exceeds failure threshold in keyspace datacenter.");
+
     /**
      * Guardrail on passwords for CREATE / ALTER ROLE statements.
      */
@@ -1597,6 +1610,18 @@ public void setDataDiskUsageMaxDiskSize(@Nullable String size)
         DEFAULT_CONFIG.setDataDiskUsageMaxDiskSize(sizeFromString(size));
     }
 
+    @Override
+    public boolean getDataDiskUsageKeyspaceWideProtectionEnabled()
+    {
+        return DEFAULT_CONFIG.getDataDiskUsageKeyspaceWideProtectionEnabled();
+    }
+
+    @Override
+    public void setDataDiskUsageKeyspaceWideProtectionEnabled(boolean enabled)
+    {
+        DEFAULT_CONFIG.setDataDiskUsageKeyspaceWideProtectionEnabled(enabled);
+    }
+
     @Override
     public int getMinimumReplicationFactorWarnThreshold()
     {

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

@@ -874,6 +874,17 @@ public interface GuardrailsMBean
     @Nullable
     String getDataDiskUsageMaxDiskSize();
 
+    /**
+     * @return Return whether a single node replicating a given keyspace being full should block writes for the
+     * entire keyspace. Returns true if this behavior is set, false otherwise.
+     */
+    boolean getDataDiskUsageKeyspaceWideProtectionEnabled();
+
+    /**
+     * @param enabled Enables or disables blocking writes for a keyspace if a node replicating that keyspace is full.
+     */
+    void setDataDiskUsageKeyspaceWideProtectionEnabled(boolean enabled);
+
     /**
      * @param size The max disk size of the data directories when calculating disk usage thresholds, as a string
      *             formatted as in, for example, {@code 10GiB}, {@code 20MiB}, {@code 30KiB} or {@code 40B}.

src/java/org/apache/cassandra/service/disk/usage/DiskUsageBroadcaster.java

@@ -18,6 +18,7 @@
 
 package org.apache.cassandra.service.disk.usage;
 
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.TimeUnit;
@@ -27,13 +28,16 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.gms.ApplicationState;
 import org.apache.cassandra.gms.EndpointState;
 import org.apache.cassandra.gms.Gossiper;
 import org.apache.cassandra.gms.IEndpointStateChangeSubscriber;
 import org.apache.cassandra.gms.VersionedValue;
 import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.Locator;
 import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.tcm.membership.Location;
 import org.apache.cassandra.utils.NoSpamLogger;
 
 /**
@@ -50,6 +54,8 @@ public class DiskUsageBroadcaster implements IEndpointStateChangeSubscriber
     private final DiskUsageMonitor monitor;
     private final ConcurrentMap<InetAddressAndPort, DiskUsageState> usageInfo = new ConcurrentHashMap<>();
     private volatile boolean hasStuffedOrFullNode = false;
+    private final ConcurrentMap<String, Set<InetAddressAndPort>> fullNodesByDatacenter = new ConcurrentHashMap<>();
+    private final ConcurrentMap<String, Set<InetAddressAndPort>> stuffedNodesByDatacenter = new ConcurrentHashMap<>();
 
     @VisibleForTesting
     public DiskUsageBroadcaster(DiskUsageMonitor monitor)
@@ -83,6 +89,34 @@ public boolean isStuffed(InetAddressAndPort endpoint)
         return state(endpoint).isStuffed();
     }
 
+    /**
+     * @return {@code true} if there exists any node in the datacenter of {@code endpoint} which has FULL disk usage.
+     */
+    @VisibleForTesting
+    public boolean isDatacenterFull(String datacenter)
+    {
+        if (!hasStuffedOrFullNode())
+        {
+            return false;
+        }
+        Set<InetAddressAndPort> fullNodes = fullNodesByDatacenter.get(datacenter);
+        return fullNodes != null && !fullNodes.isEmpty();
+    }
+
+    /**
+     * @return {@code true} if there exists any node in the datacenter of {@code endpoint} which has FULL disk usage
+     */
+    @VisibleForTesting
+    public boolean isDatacenterStuffed(String datacenter)
+    {
+        if (!hasStuffedOrFullNode())
+        {
+            return false;
+        }
+        Set<InetAddressAndPort> stuffedNodes = stuffedNodesByDatacenter.get(datacenter);
+        return stuffedNodes != null && !stuffedNodes.isEmpty();
+    }
+
     @VisibleForTesting
     public DiskUsageState state(InetAddressAndPort endpoint)
     {
@@ -114,8 +148,9 @@ public void onChange(InetAddressAndPort endpoint, ApplicationState state, Versio
             noSpamLogger.warn(String.format("Found unknown DiskUsageState: %s. Using default state %s instead.",
                                             value.value, usageState));
         }
-        usageInfo.put(endpoint, usageState);
 
+        computeUsageStateForEpDatacenter(endpoint, usageState);
+        usageInfo.put(endpoint, usageState);
         hasStuffedOrFullNode = usageState.isStuffedOrFull() || computeHasStuffedOrFullNode();
     }
 
@@ -131,6 +166,85 @@ private boolean computeHasStuffedOrFullNode()
         return false;
     }
 
+    /**
+     * Update the set of full nodes by datacenter based on the disk usage state for the given endpoint.
+     * If the node is FULL, add it to the set for its datacenter. Otherwise, remove it from the set.
+     * This method is idempotent - adding an already-present node or removing an absent node has no effect.
+     *
+     * @param endpoint   The endpoint whose state has changed.
+     * @param usageState The new disk usage state value.
+     */
+    private void computeUsageStateForEpDatacenter(InetAddressAndPort endpoint, DiskUsageState usageState)
+    {
+        Location location = location(endpoint);
+        if (location.equals(Location.UNKNOWN))
+        {
+            noSpamLogger.warn("Unable to track disk usage by datacenter for endpoint {} because we are unable to determine its location.",
+                         endpoint);
+            return;
+        }
+
+        String datacenter = location.datacenter;
+        if (usageState.isFull())
+        {
+            // Add this node to the set of full nodes for its datacenter and remove it from the stuffed nodes
+            // if it was there.
+            fullNodesByDatacenter.computeIfAbsent(datacenter, dc -> ConcurrentHashMap.newKeySet())
+                                 .add(endpoint);
+            noSpamLogger.debug("Endpoint {} is FULL, added to full nodes set for datacenter {}", endpoint, datacenter);
+            Set<InetAddressAndPort> stuffedNodes = stuffedNodesByDatacenter.get(datacenter);
+            if (stuffedNodes != null && stuffedNodes.remove(endpoint))
+            {
+                noSpamLogger.debug("Endpoint {} is now FULL. Removed it from the stuffed nodes set for datacenter {}",
+                             endpoint, datacenter);
+            }
+        }
+        else if (usageState.isStuffed())
+        {
+            // Add this node to the set of stuffed nodes for its datacenter and remove it from the full nodes
+            // if it was there.
+            stuffedNodesByDatacenter.computeIfAbsent(datacenter, dc -> ConcurrentHashMap.newKeySet())
+                                    .add(endpoint);
+            noSpamLogger.debug("Endpoint {} is now STUFFED. Added it to the stuffed nodes set for datacenter {}",
+                         endpoint, datacenter);
+            Set<InetAddressAndPort> fullNodes = fullNodesByDatacenter.get(datacenter);
+            if (fullNodes != null && fullNodes.remove(endpoint))
+            {
+                noSpamLogger.debug("Endpoint {} is now STUFFED. Removed it from full nodes set for datacenter {}",
+                             endpoint, datacenter);
+            }
+        }
+        else
+        {
+            // Remove this node from the set of full nodes and set of stuffed nodes for its datacenter if it was there.
+            Set<InetAddressAndPort> fullNodes = fullNodesByDatacenter.get(datacenter);
+            if (fullNodes != null && fullNodes.remove(endpoint))
+            {
+                noSpamLogger.debug("Endpoint {} is no longer STUFFED or FULL, removed from stuffed for datacenter {}",
+                             endpoint, datacenter);
+            }
+            Set<InetAddressAndPort> stuffedNodes = stuffedNodesByDatacenter.get(datacenter);
+            if (stuffedNodes != null && stuffedNodes.remove(endpoint))
+            {
+                noSpamLogger.debug("Endpoint {} is no longer STUFFED, removed from the stuffed set for datacenter {}",
+                             endpoint, datacenter);
+            }
+        }
+    }
+
+    private Location location(InetAddressAndPort endpoint)
+    {
+        Locator locator = DatabaseDescriptor.getLocator();
+        if (locator == null)
+        {
+            noSpamLogger.warn("Unable to track disk usage by datacenter for endpoint {} because locator is null",
+                              endpoint);
+            return Location.UNKNOWN;
+        }
+        Location location = locator.location(endpoint);
+        return location != null ? location : Location.UNKNOWN;
+    }
+
     @Override
     public void onJoin(InetAddressAndPort endpoint, EndpointState epState)
     {
@@ -164,10 +278,34 @@ public void onRestart(InetAddressAndPort endpoint, EndpointState state)
     @Override
     public void onRemove(InetAddressAndPort endpoint)
     {
+        updateDiskUsageStateForDatacenterOnRemoval(endpoint);
         usageInfo.remove(endpoint);
         hasStuffedOrFullNode = usageInfo.values().stream().anyMatch(DiskUsageState::isStuffedOrFull);
     }
 
+    private void updateDiskUsageStateForDatacenterOnRemoval(InetAddressAndPort endpoint)
+    {
+        Location nodeLocation = location(endpoint);
+        if (nodeLocation.equals(Location.UNKNOWN))
+        {
+            logger.debug("Unable to determine location for removed endpoint {}. Will not update datacenter tracking.", endpoint);
+            return;
+        }
+
+        String datacenter = nodeLocation.datacenter;
+        // Remove the endpoint from the full nodes and stuffed nodes set for its datacenter
+        Set<InetAddressAndPort> fullNodes = fullNodesByDatacenter.get(datacenter);
+        if (fullNodes != null && fullNodes.remove(endpoint))
+        {
+            logger.debug("Removed endpoint {} from full nodes set for datacenter {} on node removal", endpoint, datacenter);
+        }
+        Set<InetAddressAndPort> stuffedNodes = stuffedNodesByDatacenter.get(datacenter);
+        if (stuffedNodes != null && stuffedNodes.remove(endpoint))
+        {
+            logger.debug("Removed endpoint {} from stuffed nodes set for datacenter {} on node removal", endpoint, datacenter);
+        }
+    }
+
     private void updateDiskUsage(InetAddressAndPort endpoint, EndpointState state)
     {
         VersionedValue localValue = state.getApplicationState(ApplicationState.DISK_USAGE);

src/java/org/apache/cassandra/utils/NoSpamLogger.java

@@ -235,6 +235,16 @@ public static NoSpamLogger wrap(Logger wrapped, long minInterval, TimeUnit timeU
         return new NoSpamLogger(wrapped, minInterval, timeUnit);
     }
 
+    public boolean debug(long nowNanos, String s, Object... objects)
+    {
+        return NoSpamLogger.this.log( Level.DEBUG, s, nowNanos, objects);
+    }
+
+    public boolean debug(String s, Object... objects)
+    {
+        return NoSpamLogger.this.debug(CLOCK.nanoTime(), s, objects);
+    }
+
     public boolean info(long nowNanos, String s, Object... objects)
     {
         return NoSpamLogger.this.log( Level.INFO, s, nowNanos, objects);