Skip to content

Commit

Permalink
Add soft/hard limits to local reads to protect against reading too mu…
Browse files Browse the repository at this point in the history
…ch data in a single query

patch by David Capwell; reviewed by Caleb Rackliffe and Marcus Eriksson for CASSANDRA-16896
  • Loading branch information
dcapwell authored and maedhroz committed Sep 28, 2021
1 parent ce11fee commit c7526f9
Show file tree
Hide file tree
Showing 66 changed files with 2,813 additions and 570 deletions.
1 change: 1 addition & 0 deletions CHANGES.txt
@@ -1,4 +1,5 @@
4.1
* Add soft/hard limits to local reads to protect against reading too much data in a single query (CASSANDRA-16896)
* Avoid token cache invalidation for removing a non-member node (CASSANDRA-15290)
* Allow configuration of consistency levels on auth operations (CASSANDRA-12988)
* Add number of sstables in a compaction to compactionstats output (CASSANDRA-16844)
Expand Down
11 changes: 7 additions & 4 deletions NEWS.txt
Expand Up @@ -40,11 +40,14 @@ New features
------------
- Warn/abort thresholds added to read queries notifying clients when these thresholds trigger (by
emitting a client warning or aborting the query). This feature is disabled by default, scheduled
to be enabled in 4.2; it is controlled with the configuration client_track_warnings_enabled,
to be enabled in 4.2; it is controlled with the configuration track_warnings.enabled,
setting to true will enable this feature. Each check has its own warn/abort thresholds, currently
tombstones (tombstone_warn_threshold, and tombstone_failure_threshold) and coordinator result set
materialized size (client_large_read_warn_threshold_kb, and client_large_read_abort_threshold_kb)
are supported; more checks will be added over time.
tombstones (tombstone_warn_threshold, and tombstone_failure_threshold), coordinator result set
materialized size (track_warnings.coordinator_large_read.warn_threshold_kb, and
track_warnings.coordinator_large_read.abort_threshold_kb), local read materialized heap size
(track_warnings.local_read_size.warn_threshold_kb and track_warnings.local_read_size.abort_threshold_kb),
and RowIndexEntry estimated memory size (track_warnings.row_index_size.warn_threshold_kb and
track_warnings.row_index_size.abort_threshold_kb) are supported; more checks will be added over time.

Upgrading
---------
Expand Down
2 changes: 2 additions & 0 deletions build.xml
Expand Up @@ -859,6 +859,7 @@
<pathelement location="${test.conf}"/>
</classpath>
<jvmarg value="-Dstorage-config=${test.conf}"/>
<jvmarg value="-Dcassandra.track_warnings.coordinator.defensive_checks_enabled=true" /> <!-- enable defensive checks -->
<jvmarg value="-javaagent:${build.lib}/jamm-${jamm.version}.jar" />
<jvmarg value="-ea"/>
<jvmarg line="${java11-jvmargs}"/>
Expand Down Expand Up @@ -1370,6 +1371,7 @@
<jvmarg value="-Dcassandra.testtag=@{testtag}"/>
<jvmarg value="-Dcassandra.keepBriefBrief=${cassandra.keepBriefBrief}" />
<jvmarg value="-Dcassandra.strict.runtime.checks=true" />
<jvmarg value="-Dcassandra.track_warnings.coordinator.defensive_checks_enabled=true" /> <!-- enable defensive checks -->
<jvmarg line="${java11-jvmargs}"/>
<!-- disable shrinks in quicktheories CASSANDRA-15554 -->
<jvmarg value="-DQT_SHRINKS=0"/>
Expand Down
32 changes: 21 additions & 11 deletions conf/cassandra.yaml
Expand Up @@ -1459,15 +1459,25 @@ enable_drop_compact_storage: false
# - 127.0.0.0/31

# Enables tracking warnings/aborts across all replicas for reporting back to client.
# Scheduled to enable in 4.2
# See: CASSANDRA-16850
# See: tombstone_warn_threshold, tombstone_failure_threshold, client_large_read_warn_threshold_kb, and client_large_read_abort_threshold_kb
#client_track_warnings_enabled: false

# When client_track_warnings_enabled: true, this tracks the materialized size of a query on the
# coordinator. If client_large_read_warn_threshold_kb is greater than 0, this will emit a warning
# to clients with details on what query triggered this as well as the size of the result set; if
# client_large_read_abort_threshold_kb is greater than 0, this will abort the query after it
# has exceeded this threshold, returning a read error to the user.
#client_large_read_warn_threshold_kb: 0
#client_large_read_abort_threshold_kb: 0
#track_warnings:
# # Scheduled to enable in 4.2
# enabled: false
# # When track_warnings.enabled: true, this tracks the materialized size of a query on the
# # coordinator. If coordinator_large_read.warn_threshold_kb is greater than 0, this will emit a warning
# # to clients with details on what query triggered this as well as the size of the result set; if
# # coordinator_large_read.abort_threshold_kb is greater than 0, this will abort the query after it
# # has exceeded this threshold, returning a read error to the user.
# coordinator_large_read:
# warn_threshold_kb: 0
# abort_threshold_kb: 0
# # When track_warnings.enabled: true, this tracks the size of the local read (as defined by
# # heap size), and will warn/abort based off these thresholds; 0 disables these checks.
# local_read_size:
# warn_threshold_kb: 0
# abort_threshold_kb: 0
# # When track_warnings.enabled: true, this tracks the expected memory size of the RowIndexEntry
# # and will warn/abort based off these thresholds; 0 disables these checks.
# row_index_size:
# warn_threshold_kb: 0
# abort_threshold_kb: 0
6 changes: 3 additions & 3 deletions ide/idea/workspace.xml
Expand Up @@ -143,7 +143,7 @@
<configuration default="true" type="Application" factoryName="Application">
<extension name="coverage" enabled="false" merge="false" sample_coverage="true" runner="idea" />
<option name="MAIN_CLASS_NAME" value="" />
<option name="VM_PARAMETERS" value="-Dcassandra.config=file://$PROJECT_DIR$/conf/cassandra.yaml -Dcassandra.storagedir=$PROJECT_DIR$/data -Dlogback.configurationFile=file://$PROJECT_DIR$/conf/logback.xml -Dcassandra.logdir=$PROJECT_DIR$/data/logs -Djava.library.path=$PROJECT_DIR$/lib/sigar-bin -DQT_SHRINKS=0 -ea" />
<option name="VM_PARAMETERS" value="-Dcassandra.config=file://$PROJECT_DIR$/conf/cassandra.yaml -Dcassandra.storagedir=$PROJECT_DIR$/data -Dlogback.configurationFile=file://$PROJECT_DIR$/conf/logback.xml -Dcassandra.logdir=$PROJECT_DIR$/data/logs -Djava.library.path=$PROJECT_DIR$/lib/sigar-bin -DQT_SHRINKS=0 -ea -Dcassandra.track_warnings.coordinator.defensive_checks_enabled=true" />
<option name="PROGRAM_PARAMETERS" value="" />
<option name="WORKING_DIRECTORY" value="" />
<option name="ALTERNATIVE_JRE_PATH_ENABLED" value="false" />
Expand All @@ -167,7 +167,7 @@
<option name="MAIN_CLASS_NAME" value="" />
<option name="METHOD_NAME" value="" />
<option name="TEST_OBJECT" value="class" />
<option name="VM_PARAMETERS" value="-Dcassandra.config=file://$PROJECT_DIR$/test/conf/cassandra.yaml -Dlogback.configurationFile=file://$PROJECT_DIR$/test/conf/logback-test.xml -Dcassandra.logdir=$PROJECT_DIR$/build/test/logs -Djava.library.path=$PROJECT_DIR$/lib/sigar-bin -Dlegacy-sstable-root=$PROJECT_DIR$/test/data/legacy-sstables -Dinvalid-legacy-sstable-root=$PROJECT_DIR$/test/data/invalid-legacy-sstables -Dcassandra.ring_delay_ms=1000 -Dcassandra.skip_sync=true -ea -XX:MaxMetaspaceSize=384M -XX:SoftRefLRUPolicyMSPerMB=0 -Dcassandra.strict.runtime.checks=true -Dlegacy-sstable-root=$PROJECT_DIR$/test/data/legacy-sstables -Dinvalid-legacy-sstable-root=$PROJECT_DIR$/test/data/invalid-legacy-sstables -Dmigration-sstable-root=$PROJECT_DIR$/test/data/migration-sstables -Dcassandra.ring_delay_ms=1000 -Dcassandra.tolerate_sstable_size=true -Dcassandra.skip_sync=true" />
<option name="VM_PARAMETERS" value="-Dcassandra.config=file://$PROJECT_DIR$/test/conf/cassandra.yaml -Dlogback.configurationFile=file://$PROJECT_DIR$/test/conf/logback-test.xml -Dcassandra.logdir=$PROJECT_DIR$/build/test/logs -Djava.library.path=$PROJECT_DIR$/lib/sigar-bin -Dlegacy-sstable-root=$PROJECT_DIR$/test/data/legacy-sstables -Dinvalid-legacy-sstable-root=$PROJECT_DIR$/test/data/invalid-legacy-sstables -Dcassandra.ring_delay_ms=1000 -Dcassandra.skip_sync=true -ea -XX:MaxMetaspaceSize=384M -XX:SoftRefLRUPolicyMSPerMB=0 -Dcassandra.strict.runtime.checks=true -Dlegacy-sstable-root=$PROJECT_DIR$/test/data/legacy-sstables -Dinvalid-legacy-sstable-root=$PROJECT_DIR$/test/data/invalid-legacy-sstables -Dmigration-sstable-root=$PROJECT_DIR$/test/data/migration-sstables -Dcassandra.ring_delay_ms=1000 -Dcassandra.tolerate_sstable_size=true -Dcassandra.skip_sync=true -Dcassandra.track_warnings.coordinator.defensive_checks_enabled=true" />
<option name="PARAMETERS" value="" />
<fork_mode value="class" />
<option name="WORKING_DIRECTORY" value="" />
Expand All @@ -186,7 +186,7 @@
<configuration default="false" name="Cassandra" type="Application" factoryName="Application">
<extension name="coverage" enabled="false" merge="false" sample_coverage="true" runner="idea" />
<option name="MAIN_CLASS_NAME" value="org.apache.cassandra.service.CassandraDaemon" />
<option name="VM_PARAMETERS" value="-Dcassandra-foreground=yes -Dcassandra.config=file://$PROJECT_DIR$/conf/cassandra.yaml -Dcassandra.storagedir=$PROJECT_DIR$/data -Dlogback.configurationFile=file://$PROJECT_DIR$/conf/logback.xml -Dcassandra.logdir=$PROJECT_DIR$/data/logs -Djava.library.path=$PROJECT_DIR$/lib/sigar-bin -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=7199 -Dcom.sun.management.jmxremote.local.only=false -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -ea -Xmx1G" />
<option name="VM_PARAMETERS" value="-Dcassandra-foreground=yes -Dcassandra.config=file://$PROJECT_DIR$/conf/cassandra.yaml -Dcassandra.storagedir=$PROJECT_DIR$/data -Dlogback.configurationFile=file://$PROJECT_DIR$/conf/logback.xml -Dcassandra.logdir=$PROJECT_DIR$/data/logs -Djava.library.path=$PROJECT_DIR$/lib/sigar-bin -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=7199 -Dcom.sun.management.jmxremote.local.only=false -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -ea -Xmx1G -Dcassandra.track_warnings.coordinator.defensive_checks_enabled=true" />
<option name="PROGRAM_PARAMETERS" value="" />
<option name="WORKING_DIRECTORY" value="file://$PROJECT_DIR$" />
<option name="ALTERNATIVE_JRE_PATH_ENABLED" value="false" />
Expand Down
7 changes: 2 additions & 5 deletions src/java/org/apache/cassandra/config/Config.java
Expand Up @@ -346,19 +346,16 @@ public class Config

public MemtableAllocationType memtable_allocation_type = MemtableAllocationType.heap_buffers;

public final TrackWarnings track_warnings = new TrackWarnings();

public volatile int tombstone_warn_threshold = 1000;
public volatile int tombstone_failure_threshold = 100000;

public volatile long client_large_read_warn_threshold_kb = 0;
public volatile long client_large_read_abort_threshold_kb = 0;

public final ReplicaFilteringProtectionOptions replica_filtering_protection = new ReplicaFilteringProtectionOptions();

public volatile Long index_summary_capacity_in_mb;
public volatile int index_summary_resize_interval_in_minutes = 60;

public volatile boolean client_track_warnings_enabled = false; // should set to true in 4.2

public int gc_log_threshold_in_ms = 200;
public int gc_warn_threshold_in_ms = 1000;

Expand Down
74 changes: 59 additions & 15 deletions src/java/org/apache/cassandra/config/DatabaseDescriptor.java
Expand Up @@ -688,6 +688,7 @@ else if (conf.repair_session_space_in_mb > (int) (Runtime.getRuntime().maxMemory

applyConcurrentValidations(conf);
applyRepairCommandPoolSize(conf);
applyTrackWarningsValidations(conf);

if (conf.concurrent_materialized_view_builders <= 0)
throw new ConfigurationException("concurrent_materialized_view_builders should be strictly greater than 0, but was " + conf.concurrent_materialized_view_builders, false);
Expand Down Expand Up @@ -857,9 +858,6 @@ else if (config.concurrent_validations > config.concurrent_compactors && !allowU
throw new ConfigurationException("To set concurrent_validations > concurrent_compactors, " +
"set the system property cassandra.allow_unlimited_concurrent_validations=true");
}

conf.client_large_read_warn_threshold_kb = Math.max(conf.client_large_read_warn_threshold_kb, 0);
conf.client_large_read_abort_threshold_kb = Math.max(conf.client_large_read_abort_threshold_kb, 0);
}

@VisibleForTesting
Expand All @@ -869,6 +867,12 @@ static void applyRepairCommandPoolSize(Config config)
config.repair_command_pool_size = config.concurrent_validations;
}

@VisibleForTesting
static void applyTrackWarningsValidations(Config config)
{
config.track_warnings.validate("track_warnings");
}

private static String storagedirFor(String type)
{
return storagedir(type + "_directory") + File.separator + type;
Expand Down Expand Up @@ -3477,33 +3481,73 @@ public static SubnetGroups getInternodeErrorReportingExclusions()
return conf.internode_error_reporting_exclusions;
}

public static long getClientLargeReadWarnThresholdKB()
public static boolean getTrackWarningsEnabled()
{
return conf.track_warnings.enabled;
}

public static void setTrackWarningsEnabled(boolean value)
{
conf.track_warnings.enabled = value;
}

public static long getCoordinatorReadSizeWarnThresholdKB()
{
return conf.track_warnings.coordinator_read_size.getWarnThresholdKb();
}

public static void setCoordinatorReadSizeWarnThresholdKB(long threshold)
{
conf.track_warnings.coordinator_read_size.setWarnThresholdKb(threshold);
}

public static long getCoordinatorReadSizeAbortThresholdKB()
{
return conf.track_warnings.coordinator_read_size.getAbortThresholdKb();
}

public static void setCoordinatorReadSizeAbortThresholdKB(long threshold)
{
conf.track_warnings.coordinator_read_size.setAbortThresholdKb(threshold);
}

public static long getLocalReadSizeWarnThresholdKb()
{
return conf.track_warnings.local_read_size.getWarnThresholdKb();
}

public static void setLocalReadSizeWarnThresholdKb(long value)
{
conf.track_warnings.local_read_size.setWarnThresholdKb(value);
}

public static long getLocalReadSizeAbortThresholdKb()
{
return conf.client_large_read_warn_threshold_kb;
return conf.track_warnings.local_read_size.getAbortThresholdKb();
}

public static void setClientLargeReadWarnThresholdKB(long threshold)
public static void setLocalReadSizeAbortThresholdKb(long value)
{
conf.client_large_read_warn_threshold_kb = Math.max(threshold, 0);
conf.track_warnings.local_read_size.setAbortThresholdKb(value);
}

public static long getClientLargeReadAbortThresholdKB()
public static int getRowIndexSizeWarnThresholdKb()
{
return conf.client_large_read_abort_threshold_kb;
return conf.track_warnings.row_index_size.getWarnThresholdKb();
}

public static void setClientLargeReadAbortThresholdKB(long threshold)
public static void setRowIndexSizeWarnThresholdKb(int value)
{
conf.client_large_read_abort_threshold_kb = Math.max(threshold, 0);
conf.track_warnings.row_index_size.setWarnThresholdKb(value);
}

public static boolean getClientTrackWarningsEnabled()
public static int getRowIndexSizeAbortThresholdKb()
{
return conf.client_track_warnings_enabled;
return conf.track_warnings.row_index_size.getAbortThresholdKb();
}

public static void setClientTrackWarningsEnabled(boolean value)
public static void setRowIndexSizeAbortThresholdKb(int value)
{
conf.client_track_warnings_enabled = value;
conf.track_warnings.row_index_size.setAbortThresholdKb(value);
}
}
108 changes: 108 additions & 0 deletions src/java/org/apache/cassandra/config/TrackWarnings.java
@@ -0,0 +1,108 @@
/*
* 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.config;

import org.apache.cassandra.exceptions.ConfigurationException;

public class TrackWarnings
{
public volatile boolean enabled = false; // should set to true in 4.2
public final LongByteThreshold coordinator_read_size = new LongByteThreshold();
public final LongByteThreshold local_read_size = new LongByteThreshold();
public final IntByteThreshold row_index_size = new IntByteThreshold();

public void validate(String prefix)
{
prefix += ".";
coordinator_read_size.validate(prefix + "coordinator_read_size");
local_read_size.validate(prefix + "local_read_size");
row_index_size.validate(prefix + "row_index_size");
}

public static class LongByteThreshold
{
public volatile long warn_threshold_kb = 0;
public volatile long abort_threshold_kb = 0;

public long getWarnThresholdKb()
{
return warn_threshold_kb;
}

public void setWarnThresholdKb(long value)
{
warn_threshold_kb = Math.max(value, 0);
}

public long getAbortThresholdKb()
{
return abort_threshold_kb;
}

public void setAbortThresholdKb(long value)
{
abort_threshold_kb = Math.max(value, 0);
}

public void validate(String prefix)
{
warn_threshold_kb = Math.max(warn_threshold_kb, 0);
abort_threshold_kb = Math.max(abort_threshold_kb, 0);

if (abort_threshold_kb != 0 && abort_threshold_kb < warn_threshold_kb)
throw new ConfigurationException(String.format("abort_threshold_kb (%d) must be greater than or equal to warn_threshold_kb (%d); see %s",
abort_threshold_kb, warn_threshold_kb, prefix));
}
}

public static class IntByteThreshold
{
public volatile int warn_threshold_kb = 0;
public volatile int abort_threshold_kb = 0;

public int getWarnThresholdKb()
{
return warn_threshold_kb;
}

public void setWarnThresholdKb(int value)
{
warn_threshold_kb = Math.max(value, 0);
}

public int getAbortThresholdKb()
{
return abort_threshold_kb;
}

public void setAbortThresholdKb(int value)
{
abort_threshold_kb = Math.max(value, 0);
}

public void validate(String prefix)
{
warn_threshold_kb = Math.max(warn_threshold_kb, 0);
abort_threshold_kb = Math.max(abort_threshold_kb, 0);

if (abort_threshold_kb != 0 && abort_threshold_kb < warn_threshold_kb)
throw new ConfigurationException(String.format("abort_threshold_kb (%d) must be greater than or equal to warn_threshold_kb (%d); see %s",
abort_threshold_kb, warn_threshold_kb, prefix));
}
}
}

0 comments on commit c7526f9

Please sign in to comment.