natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add guardrail for query page size 

patch by Bartlomiej; reviewed by Andrés de la Peña and Brandon Williams for CASSANDRA-17189
This commit is contained in:
Kowalczyk 2021-12-11 13:17:39 +00:00 committed by Andrés de la Peña
parent 75482d0a8c
commit bf1446cd85
13 changed files with 243 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.1
* Add guardrail for query page size (CASSANDRA-17189)
* Allow column_index_size_in_kb to be configurable through nodetool (CASSANDRA-17121)
* Emit a metric for number of local read and write calls
* Add non-blocking mode for CDC writes (CASSANDRA-17001)

5
conf/cassandra.yaml
View File

@ -1607,3 +1607,8 @@ enable_drop_compact_storage: false
# disallowed: []
# Guardrail to allow/disallow user-provided timestamps. Defaults to true.
# user_timestamps_enabled: true
# Guardrail to warn or abort when using a page size greater than threshold.
# The two thresholds default to -1 to disable.
# page_size:
# warn_threshold: -1
# abort_threshold: -1

9
src/java/org/apache/cassandra/config/GuardrailsOptions.java
View File

@ -57,6 +57,8 @@ public class GuardrailsOptions implements GuardrailsConfig
public final IntThreshold secondary_indexes_per_table = new IntThreshold();
public final IntThreshold materialized_views_per_table = new IntThreshold();
public final TableProperties table_properties = new TableProperties();
public final IntThreshold page_size = new IntThreshold();
public volatile boolean user_timestamps_enabled = true;
public void validate()
@ -66,6 +68,7 @@ public class GuardrailsOptions implements GuardrailsConfig
secondary_indexes_per_table.validate("guardrails.secondary_indexes_per_table");
materialized_views_per_table.validate("guardrails.materialized_views_per_table");
table_properties.validate("guardrails.table_properties");
page_size.validate("guardrails.page_size");
}
@Override
@ -120,6 +123,12 @@ public class GuardrailsOptions implements GuardrailsConfig
return user_timestamps_enabled;
}
@Override
public IntThreshold getPageSize()
{
return page_size;
}
public void setUserTimestampsEnabled(boolean enabled)
{
user_timestamps_enabled = enabled;

4
src/java/org/apache/cassandra/cql3/QueryProcessor.java
View File

@ -763,13 +763,13 @@ public class QueryProcessor implements QueryHandler
{
ModificationStatement modificationStatement = ((ModificationStatement) statement);
statementKsName = modificationStatement.keyspace();
statementCfName = modificationStatement.columnFamily();
statementCfName = modificationStatement.table();
}
else if (statement instanceof SelectStatement)
{
SelectStatement selectStatement = ((SelectStatement) statement);
statementKsName = selectStatement.keyspace();
statementCfName = selectStatement.columnFamily();
statementCfName = selectStatement.table();
}
else if (statement instanceof BatchStatement)
{

4
src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
View File

@ -233,10 +233,10 @@ public class BatchStatement implements CQLStatement
String cfName = null;
for (ModificationStatement stmt : statements)
{
if (ksName != null && (!stmt.keyspace().equals(ksName) || !stmt.columnFamily().equals(cfName)))
if (ksName != null && (!stmt.keyspace().equals(ksName) || !stmt.table().equals(cfName)))
throw new InvalidRequestException("Batch with conditions cannot span multiple tables");
ksName = stmt.keyspace();
cfName = stmt.columnFamily();
cfName = stmt.table();
}
}
}

2
src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java
View File

@ -198,6 +198,6 @@ public class DeleteStatement extends ModificationStatement
@Override
public AuditLogContext getAuditLogContext()
{
return new AuditLogContext(AuditLogEntryType.DELETE, keyspace(), columnFamily());
return new AuditLogContext(AuditLogEntryType.DELETE, keyspace(), table());
}
}

8
src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
View File

@ -202,7 +202,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
return metadata.keyspace;
}
public String columnFamily()
public String table()
{
return metadata.name;
}
@ -247,7 +247,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
// MV updates need to get the current state from the table, and might update the views
// Require Permission.SELECT on the base table, and Permission.MODIFY on the views
Iterator<ViewMetadata> views = View.findAll(keyspace(), columnFamily()).iterator();
Iterator<ViewMetadata> views = View.findAll(keyspace(), table()).iterator();
if (views.hasNext())
{
state.ensureTablePermission(metadata, Permission.SELECT);
@ -492,7 +492,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
CQL3CasRequest request = makeCasRequest(queryState, options);
try (RowIterator result = StorageProxy.cas(keyspace(),
columnFamily(),
table(),
request.key,
request,
options.getSerialConsistency(),
@ -550,7 +550,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
private ResultSet buildCasResultSet(RowIterator partition, QueryState state, QueryOptions options)
{
return buildCasResultSet(keyspace(), columnFamily(), partition, getColumnsWithConditions(), false, state, options);
return buildCasResultSet(keyspace(), table(), partition, getColumnsWithConditions(), false, state, options);
}
static ResultSet buildCasResultSet(String ksName,

16
src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
View File

@ -31,6 +31,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.audit.AuditLogContext;
import org.apache.cassandra.audit.AuditLogEntryType;
import org.apache.cassandra.auth.Permission;
import org.apache.cassandra.db.guardrails.Guardrails;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableMetadata;
@ -215,7 +216,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
{
if (table.isView())
{
TableMetadataRef baseTable = View.findBaseTable(keyspace(), columnFamily());
TableMetadataRef baseTable = View.findBaseTable(keyspace(), table());
if (baseTable != null)
state.ensureTablePermission(baseTable, Permission.SELECT);
}
@ -256,7 +257,8 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
QueryPager pager = getPager(query, options);
return execute(Pager.forDistributedQuery(pager, cl, state.getClientState()),
return execute(state,
Pager.forDistributedQuery(pager, cl, state.getClientState()),
options,
selectors,
pageSize,
@ -379,7 +381,8 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
}
}
private ResultMessage.Rows execute(Pager pager,
private ResultMessage.Rows execute(QueryState state,
Pager pager,
QueryOptions options,
Selectors selectors,
int pageSize,
@ -387,6 +390,8 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
int userLimit,
long queryStartNanoTime) throws RequestValidationException, RequestExecutionException
{
Guardrails.pageSize.guard(pageSize, table(), state.getClientState());
if (aggregationSpec != null)
{
if (!restrictions.hasPartitionKeyRestrictions())
@ -461,7 +466,8 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
QueryPager pager = getPager(query, options);
return execute(Pager.forInternalQuery(pager, executionController),
return execute(state,
Pager.forInternalQuery(pager, executionController),
options,
selectors,
pageSize,
@ -494,7 +500,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
return table.keyspace;
}
public String columnFamily()
public String table()
{
return table.name;
}

2
src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
View File

@ -334,6 +334,6 @@ public class UpdateStatement extends ModificationStatement
@Override
public AuditLogContext getAuditLogContext()
{
return new AuditLogContext(AuditLogEntryType.UPDATE, keyspace(), columnFamily());
return new AuditLogContext(AuditLogEntryType.UPDATE, keyspace(), table());
}
}

30
src/java/org/apache/cassandra/db/guardrails/Guardrails.java
View File

@ -98,6 +98,18 @@ public final class Guardrails implements GuardrailsMBean
new DisableFlag(state -> !CONFIG_PROVIDER.getOrCreate(state).getUserTimestampsEnabled(),
"User provided timestamps (USING TIMESTAMP)");
/**
* Guardrail on the number of elements returned within page.
*/
public static final Threshold pageSize =
new Threshold(state -> CONFIG_PROVIDER.getOrCreate(state).getPageSize(),
(isWarning, what, value, threshold) ->
isWarning ? format("Query for table %s with page size %s exceeds warning threshold of %s.",
what, value, threshold)
: format("Aborting query for table %s, page size %s exceeds abort threshold of %s.",
what, value, threshold));
private Guardrails()
{
MBeanWrapper.instance.registerMBean(this, MBEAN_NAME);
@ -268,6 +280,24 @@ public final class Guardrails implements GuardrailsMBean
DEFAULT_CONFIG.setUserTimestampsEnabled(enabled);
}
@Override
public int getPageSizeWarnThreshold()
{
return (int) DEFAULT_CONFIG.getPageSize().getWarnThreshold();
}
@Override
public int getPageSizeAbortThreshold()
{
return (int) DEFAULT_CONFIG.getPageSize().getAbortThreshold();
}
@Override
public void setPageSizeThreshold(int warn, int abort)
{
DEFAULT_CONFIG.getPageSize().setThresholds(warn, abort);
}
private static String toCSV(Set<String> values)
{
return values == null ? "" : String.join(",", values);

5
src/java/org/apache/cassandra/db/guardrails/GuardrailsConfig.java
View File

@ -79,4 +79,9 @@ public interface GuardrailsConfig
* @return {@code true} if user-provided timestamps are allowed, {@code false} otherwise.
*/
boolean getUserTimestampsEnabled();
/**
* @return The threshold to warn or abort when page size exceeds given size.
*/
Threshold.Config getPageSize();
}

18
src/java/org/apache/cassandra/db/guardrails/GuardrailsMBean.java
View File

@ -170,4 +170,22 @@ public interface GuardrailsMBean
* @param enabled {@code true} if user-provided timestamps are allowed, {@code false} otherwise.
*/
void setUserTimestampsEnabled(boolean enabled);
/**
* @return The threshold to warn when requested page size greater than threshold.
* -1 means disabled.
*/
int getPageSizeWarnThreshold();
/**
* @return The threshold to prevent requesting page with more elements than threshold.
* -1 means disabled.
*/
int getPageSizeAbortThreshold();
/**
* @param warn The threshold to warn when the requested page size is greater than threshold. -1 means disabled.
* @param abort The threshold to prevent requesting pages with more elements than threshold. -1 means disabled.
*/
void setPageSizeThreshold(int warn, int abort);
}

154
test/unit/org/apache/cassandra/db/guardrails/GuardrailPageSizeTest.java Normal file
View File

@ -0,0 +1,154 @@
/*
* 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.db.guardrails;
import java.util.Collections;
import org.junit.Before;
import org.junit.Test;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLStatement;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.transport.ProtocolVersion;
import static java.lang.String.format;
/**
* Tests the guardrail for the page size, {@link Guardrails#pageSize}.
*/
public class GuardrailPageSizeTest extends ThresholdTester
{
private static final int PAGE_SIZE_WARN_THRESHOLD = 5;
private static final int PAGE_SIZE_ABORT_THRESHOLD = 10;
public GuardrailPageSizeTest()
{
super(PAGE_SIZE_WARN_THRESHOLD,
PAGE_SIZE_ABORT_THRESHOLD,
DatabaseDescriptor.getGuardrailsConfig().getPageSize(),
Guardrails::setPageSizeThreshold,
Guardrails::getPageSizeWarnThreshold,
Guardrails::getPageSizeAbortThreshold);
}
@Before
public void setupTest()
{
createTable("CREATE TABLE IF NOT EXISTS %s (k INT, c INT, v TEXT, PRIMARY KEY(k, c))");
}
@Test
public void testSelectStatementAgainstPageSize() throws Throwable
{
// regular query
String query = "SELECT * FROM %s";
assertPagingValid(query, 3);
assertPagingValid(query, PAGE_SIZE_WARN_THRESHOLD);
assertPagingWarns(query, 6);
assertPagingWarns(query, PAGE_SIZE_ABORT_THRESHOLD);
assertPagingAborts(query, 11);
// aggregation query
query = "SELECT COUNT(*) FROM %s WHERE k=0";
assertPagingValid(query, 3);
assertPagingValid(query, PAGE_SIZE_WARN_THRESHOLD);
assertPagingWarns(query, 6);
assertPagingWarns(query, PAGE_SIZE_ABORT_THRESHOLD);
assertPagingAborts(query, 11);
// query with limit over thresholds
query = "SELECT * FROM %s LIMIT 100";
assertPagingValid(query, 3);
assertPagingValid(query, PAGE_SIZE_WARN_THRESHOLD);
assertPagingWarns(query, 6);
assertPagingWarns(query, PAGE_SIZE_ABORT_THRESHOLD);
assertPagingAborts(query, 11);
// query with limit under thresholds
query = "SELECT * FROM %s LIMIT 1";
assertPagingValid(query, 3);
assertPagingValid(query, PAGE_SIZE_WARN_THRESHOLD);
assertPagingValid(query, 6);
assertPagingValid(query, PAGE_SIZE_ABORT_THRESHOLD);
assertPagingValid(query, 11);
}
@Test
public void testExcludedUsers() throws Throwable
{
assertPagingIgnored("SELECT * FROM %s", PAGE_SIZE_WARN_THRESHOLD + 1);
assertPagingIgnored("SELECT * FROM %s", PAGE_SIZE_ABORT_THRESHOLD + 1);
}
private void assertPagingValid(String query, int pageSize) throws Throwable
{
assertValid(() -> executeWithPaging(userClientState, query, pageSize));
}
private void assertPagingIgnored(String query, int pageSize) throws Throwable
{
assertValid(() -> executeWithPaging(superClientState, query, pageSize));
assertValid(() -> executeWithPaging(systemClientState, query, pageSize));
}
private void assertPagingWarns(String query, int pageSize) throws Throwable
{
assertWarns(() -> executeWithPaging(userClientState, query, pageSize),
format("Query for table %s with page size %s exceeds warning threshold of %s.",
currentTable(), pageSize, PAGE_SIZE_WARN_THRESHOLD));
}
private void assertPagingAborts(String query, int pageSize) throws Throwable
{
assertAborts(() -> executeWithPaging(userClientState, query, pageSize),
format("Aborting query for table %s, page size %s exceeds abort threshold of %s.",
currentTable(), pageSize, PAGE_SIZE_ABORT_THRESHOLD));
}
private void executeWithPaging(ClientState state, String query, int pageSize)
{
QueryState queryState = new QueryState(state);
String formattedQuery = formatQuery(query);
CQLStatement statement = QueryProcessor.parseStatement(formattedQuery, queryState.getClientState());
statement.validate(state);
QueryOptions options = QueryOptions.create(ConsistencyLevel.ONE,
Collections.emptyList(),
false,
pageSize,
null,
null,
ProtocolVersion.CURRENT,
KEYSPACE);
statement.executeLocally(queryState, options);
}
//not used by page-size guardrail tests.
protected long currentValue()
{
throw new UnsupportedOperationException();
}
}