diff --git a/CHANGES.txt b/CHANGES.txt index 5b7cb6469c..1d86f30789 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 5.1 + * Implement appender of slow queries to system_views.slow_queries table (CASSANDRA-13001) * Add autocompletion in CQLSH for built-in functions (CASSANDRA-19631) * Grant permission on keyspaces system_views and system_virtual_schema not possible (CASSANDRA-20171) * General Purpose Transactions (Accord) [CEP-15] (CASSANDRA-17092) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index fe80fda164..0c68afe500 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -1396,6 +1396,10 @@ request_timeout: 10000ms # How long before a node logs slow queries. Select queries that take longer than # this timeout to execute, will generate an aggregated log message, so that slow queries # can be identified. Set this value to zero to disable slow query logging. +# +# It is possible to log slow queries into system_views.slow_queries virtual table. +# Consult logback.xml to uncomment specific appender and logger to enable this functionality. +# # Min unit: ms slow_query_log_timeout: 500ms diff --git a/conf/logback.xml b/conf/logback.xml index 102cf06352..a8dabf6564 100644 --- a/conf/logback.xml +++ b/conf/logback.xml @@ -119,6 +119,18 @@ appender reference in the root level section below. --> + + + + + + + + + + diff --git a/doc/modules/cassandra/pages/managing/configuration/cass_logback_xml_file.adoc b/doc/modules/cassandra/pages/managing/configuration/cass_logback_xml_file.adoc index a62dfe91a7..b6e4d5f545 100644 --- a/doc/modules/cassandra/pages/managing/configuration/cass_logback_xml_file.adoc +++ b/doc/modules/cassandra/pages/managing/configuration/cass_logback_xml_file.adoc @@ -80,7 +80,7 @@ Specify the format of the message. Part of the rolling policy. %-5level [%thread] %date\{ISO8601} %F:%L - %msg%n -=== Logging to Cassandra virtual table +=== Logging system logs to Cassandra virtual table It is possible to configure logback.xml in such a way that logs would appear in `system_views.system_log` table. This is achieved by appender implemented in class `VirtualTableAppender` which is called `CQLLOG` in the @@ -101,6 +101,72 @@ each message will occupy memory. The appender to virtual table is commented out by default so logging to virtual table is not active. +=== Logging slow queries to Cassandra virual table + +It is possible to log slow queries into `system_views.slow_queries` table. A query is evaluated to be slow +if it takes more than `slow_query_log_timeout` in `cassandra.yaml`. + +To log messages to `system_views.slow_queries` you need to: + +1. uncomment `SLOW_QUERIES_APPENDER` log appender +2. uncomment `appender-ref` pointing to `SLOW_QUERIES_APPENDER` in `slow_queries` logger: + +The respective configuration in `logback.xml` looks like this: + +[source,XML] +---- + + + + + + + + +---- + +By default, slow queries will be logged to `debug.log`. By uncommenting virtual table appender, it will be +logged to `debug.log` as well as to `system_views.slow_queries`. If you want to log it to `system_views.slow_queries` only, you need to comment out `DEBUGLOG` `appender-ref` in `slow_queries` logger declaration. + +If you want to log slow queries to a dedicated log file (which is e.g. rotated), that is also possible +by pointing `slow_queries` logger to a respective file appender of a given reference, similar to `DEBUGLOG` where all logs go by default. + +The structure of a table looks like this: + +[source,cql] +---- +cassandra@cqlsh> DESCRIBE system_views.slow_queries ; + +/* +Warning: Table system_views.slow_queries is a virtual table and cannot be recreated with CQL. +Structure, for reference: +VIRTUAL TABLE system_views.slow_queries ( + keyspace_name text, + table_name text, + timestamp timestamp, + query text, + avg_ms bigint, + cross_node boolean, + max_ms bigint, + min_ms bigint, + times_reported int, + PRIMARY KEY (keyspace_name, table_name, timestamp, query) +) WITH CLUSTERING ORDER BY (table_name ASC, timestamp ASC, query ASC) + AND comment = 'Slow queries'; +---- + +By having slow queries in a virtual table, an operator can check if there are slow queries for some table, see if +some queries violate some time threshold etc. The rows in this table are same data as one would get in `debug.log`, they +are just way more convenient to parse and query. + +`system_views.slow_queries` table is limited on number of rows it can hold, by default 10 000, configurable by `cassandra.virtual.slow_queries.max.rows` system property. If this table is full, the oldest entry is removed and the newest is inserted. This virtual table can be truncated by CQL and deletion on partition key (`keyspace_name` column) is allowed. + +A reader noticed that by placing custom appender implementation of `SLOW_QUERIES_APPENDER` appender on a class path and referencing it in `logback.xml`, it is possible to log slow queries wherever we have an appender for it. + === Contents of default `logback.xml` [source,XML] diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index 7459fecb8c..e2957ce95f 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -27,6 +27,7 @@ import com.google.common.primitives.Ints; import accord.utils.Invariants; import org.apache.cassandra.db.virtual.LogMessagesTable; +import org.apache.cassandra.db.virtual.SlowQueriesTable; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.StorageCompatibilityMode; @@ -363,6 +364,8 @@ public enum CassandraRelevantProperties LOG4J2_DISABLE_JMX_LEGACY("log4j2.disable.jmx"), LOG4J_SHUTDOWN_HOOK_ENABLED("log4j.shutdownHookEnabled"), LOGBACK_CONFIGURATION_FILE("logback.configurationFile"), + /** Maximum number of rows in system_views.slow_queries */ + LOGS_SLOW_QUERIES_VIRTUAL_TABLE_MAX_ROWS("cassandra.virtual.slow_queries.max.rows", convertToString(SlowQueriesTable.LOGS_VIRTUAL_TABLE_DEFAULT_ROWS)), /** Maximum number of rows in system_views.logs table */ LOGS_VIRTUAL_TABLE_MAX_ROWS("cassandra.virtual.logs.max.rows", convertToString(LogMessagesTable.LOGS_VIRTUAL_TABLE_DEFAULT_ROWS)), /** diff --git a/src/java/org/apache/cassandra/db/AbstractReadQuery.java b/src/java/org/apache/cassandra/db/AbstractReadQuery.java index 448069cfca..2e72c7ec4f 100644 --- a/src/java/org/apache/cassandra/db/AbstractReadQuery.java +++ b/src/java/org/apache/cassandra/db/AbstractReadQuery.java @@ -118,4 +118,16 @@ abstract class AbstractReadQuery extends MonitorableImpl implements ReadQuery } protected abstract void appendCQLWhereClause(StringBuilder sb); + + @Override + public String monitoredOnKeyspace() + { + return metadata().keyspace; + } + + @Override + public String monitoredOnTable() + { + return metadata().name; + } } \ No newline at end of file diff --git a/src/java/org/apache/cassandra/db/monitoring/Monitorable.java b/src/java/org/apache/cassandra/db/monitoring/Monitorable.java index 10bd10438a..4288a667de 100644 --- a/src/java/org/apache/cassandra/db/monitoring/Monitorable.java +++ b/src/java/org/apache/cassandra/db/monitoring/Monitorable.java @@ -20,6 +20,8 @@ package org.apache.cassandra.db.monitoring; public interface Monitorable { + Monitorable NO_OP = new NoOp(); + String name(); long creationTimeNanos(); long timeoutNanos(); @@ -33,4 +35,76 @@ public interface Monitorable boolean abort(); boolean complete(); + + default String monitoredOnKeyspace() { return null; }; + default String monitoredOnTable() { return null; }; + + class NoOp implements Monitorable + { + @Override + public String name() + { + return null; + } + + @Override + public long creationTimeNanos() + { + return 0; + } + + @Override + public long timeoutNanos() + { + return 0; + } + + @Override + public long slowTimeoutNanos() + { + return 0; + } + + @Override + public boolean isInProgress() + { + return false; + } + + @Override + public boolean isAborted() + { + return false; + } + + @Override + public boolean isCompleted() + { + return false; + } + + @Override + public boolean isSlow() + { + return false; + } + + @Override + public boolean isCrossNode() + { + return false; + } + + @Override + public boolean abort() + { + return false; + } + + @Override + public boolean complete() + { + return false; + } + } } diff --git a/src/java/org/apache/cassandra/db/monitoring/MonitoringTask.java b/src/java/org/apache/cassandra/db/monitoring/MonitoringTask.java index 243569910b..4d6d995c77 100644 --- a/src/java/org/apache/cassandra/db/monitoring/MonitoringTask.java +++ b/src/java/org/apache/cassandra/db/monitoring/MonitoringTask.java @@ -19,6 +19,7 @@ package org.apache.cassandra.db.monitoring; import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -32,9 +33,19 @@ import com.google.common.annotations.VisibleForTesting; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.annotation.JsonSubTypes; +import com.fasterxml.jackson.annotation.JsonTypeInfo; +import com.fasterxml.jackson.core.type.TypeReference; import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.config.CassandraRelevantProperties; +import org.apache.cassandra.utils.Clock; +import org.apache.cassandra.utils.JsonUtils; import org.apache.cassandra.utils.NoSpamLogger; +import org.apache.cassandra.utils.logging.LoggingSupport; +import org.apache.cassandra.utils.logging.LoggingSupportFactory; +import org.apache.cassandra.utils.logging.SlowQueriesAppender; import static java.util.concurrent.TimeUnit.NANOSECONDS; import static org.apache.cassandra.config.CassandraRelevantProperties.MONITORING_MAX_OPERATIONS; @@ -47,8 +58,9 @@ import static org.apache.cassandra.utils.concurrent.BlockingQueues.newBlockingQu * We also log timed out operations, see CASSANDRA-7392. * Since CASSANDRA-12403 we also log queries that were slow. */ -class MonitoringTask +public class MonitoringTask { + private static final String SLOW_OPERATIONS_LOGGER_NAME = "slow_queries"; private static final String LINE_SEPARATOR = CassandraRelevantProperties.LINE_SEPARATOR.getString(); private static final Logger logger = LoggerFactory.getLogger(MonitoringTask.class); private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 5L, TimeUnit.MINUTES); @@ -70,6 +82,8 @@ class MonitoringTask private final ScheduledFuture reportingTask; private final OperationsQueue failedOperationsQueue; private final OperationsQueue slowOperationsQueue; + private Logger slowOperationsLogger = logger; + private boolean slowOperationsLoggedToVirtualTable; private long approxLastLogTimeNanos; @@ -97,6 +111,15 @@ class MonitoringTask reportIntervalMillis, reportIntervalMillis, TimeUnit.MILLISECONDS); + + LoggingSupport support = LoggingSupportFactory.getLoggingSupport(); + if (support.getLogger(SLOW_OPERATIONS_LOGGER_NAME).isPresent()) + { + if (support.getAppender(SlowQueriesAppender.class, SlowQueriesAppender.APPENDER_NAME).isPresent()) + slowOperationsLoggedToVirtualTable = true; + + slowOperationsLogger = LoggerFactory.getLogger(SLOW_OPERATIONS_LOGGER_NAME); + } } public void cancel() @@ -169,14 +192,30 @@ class MonitoringTask if (!slowOperations.isEmpty()) { long approxElapsedNanos = approxCurrentTimeNanos - approxLastLogTimeNanos; - noSpamLogger.info("Some operations were slow, details available at debug level (debug.log)"); + noSpamLogger.info("Some operations were slow, details available at debug level (debug.log) or " + + "system_views.slow_queries virtual table (when enabled)."); + + if (slowOperationsLogger.isDebugEnabled()) + { + if (slowOperationsLoggedToVirtualTable) + { + // This is the crux of the patch for appending to vtable. + // Because we can send only Strings to debug method (or objects, on which toString() + // would be eventually called), we need to log a string in such a way that we can + // get Operation object(s) back "on the other side" when dealing with vtables and custom appenders + // as appenders work with LoggingEvent where message is just a string. + // It would be very hard / tricky / error-prone to parse customly crafted log message + // which appears in logs when no vtable appender is used. + slowOperationsLogger.debug(Operation.serialize(slowOperations.getOperations())); + } + else + slowOperationsLogger.debug("{} operations were slow in the last {} msecs:{}{}", + slowOperations.num(), + NANOSECONDS.toMillis(approxElapsedNanos), + LINE_SEPARATOR, + slowOperations.getLogMessage()); + } - if (logger.isDebugEnabled()) - logger.debug("{} operations were slow in the last {} msecs:{}{}", - slowOperations.num(), - NANOSECONDS.toMillis(approxElapsedNanos), - LINE_SEPARATOR, - slowOperations.getLogMessage()); return true; } return false; @@ -274,6 +313,12 @@ class MonitoringTask return operations.size() + numDropped; } + private Collection getOperations() + { + return operations.values(); + } + + @JsonIgnore String getLogMessage() { if (isEmpty()) @@ -307,9 +352,16 @@ class MonitoringTask * same name (CQL query text) is reported and store the average, min and max * times. */ - protected abstract static class Operation + @JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, property = "id") + @JsonSubTypes({ @JsonSubTypes.Type(value = SlowOperation.class) }) + @VisibleForTesting + public abstract static class Operation { + @JsonProperty + String id = getClass().getName(); + /** The operation that was reported as slow or timed out */ + @JsonIgnore final Monitorable operation; /** The number of times the operation was reported */ @@ -319,24 +371,50 @@ class MonitoringTask long totalTimeNanos; /** The maximum time spent by this operation */ - long maxTime; + long maxTimeNanos; /** The minimum time spent by this operation */ - long minTime; + long minTimeNanos; /** The name of the operation, i.e. the SELECT query CQL, * this is set lazily as it takes time to build the query CQL */ private String name; + /** + * creation time of this Operation object, in ms, + * this is different from operation's creationTimeNanos + * which does not follow wall clock and is useless for + * reporting purposes e.g. in virtual tables + */ + private final long timestampMs; + + // optional keyspace and table this operation acts on + // used upon deserialization + private String keyspace; + private String table; + private boolean crossNode; + Operation(Monitorable operation, long failedAtNanos) { this.operation = operation; numTimesReported = 1; totalTimeNanos = failedAtNanos - operation.creationTimeNanos(); - minTime = totalTimeNanos; - maxTime = totalTimeNanos; + minTimeNanos = totalTimeNanos; + maxTimeNanos = totalTimeNanos; + timestampMs = Clock.Global.currentTimeMillis() - (Clock.Global.nanoTime() - operation.creationTimeNanos()) / 1_000_000; } + void add(Operation operation) + { + numTimesReported++; + totalTimeNanos += operation.totalTimeNanos; + maxTimeNanos = Math.max(maxTimeNanos, operation.maxTimeNanos); + minTimeNanos = Math.min(minTimeNanos, operation.minTimeNanos); + } + + public abstract String getLogMessage(); + + @JsonProperty public String name() { if (name == null) @@ -344,15 +422,96 @@ class MonitoringTask return name; } - void add(Operation operation) + @JsonProperty + public String keyspace() { - numTimesReported++; - totalTimeNanos += operation.totalTimeNanos; - maxTime = Math.max(maxTime, operation.maxTime); - minTime = Math.min(minTime, operation.minTime); + if (operation != null) + { + String monitored = operation.monitoredOnKeyspace(); + if (monitored != null) + return monitored; + } + return keyspace; } - public abstract String getLogMessage(); + public void setKeyspace(String keyspace) + { + this.keyspace = keyspace; + } + + public void setTable(String table) + { + this.table = table; + } + + @JsonProperty + public String table() + { + if (operation != null) + { + String monitored = operation.monitoredOnTable(); + if (monitored != null) + return monitored; + } + return table; + } + + @JsonProperty + public boolean isCrossNode() + { + if (operation != null) + return operation.isCrossNode(); + + return crossNode; + } + + @JsonProperty + public int numTimesReported() + { + return numTimesReported; + } + + @JsonProperty + public long totalTimeNanos() + { + return totalTimeNanos; + } + + @JsonProperty + public long maxTimeNanos() + { + return maxTimeNanos; + } + + @JsonProperty + public long minTimeNanos() + { + return minTimeNanos; + } + + @JsonIgnore + public long averageTime() + { + return totalTimeNanos / numTimesReported; + } + + @JsonProperty + public long timestampMs() + { + return timestampMs; + } + + public static String serialize(Collection operations) + { + return JsonUtils.writeAsJsonString(operations); + } + + private static final TypeReference> TYPE_REFERENCE = new TypeReference<>() {}; + + public static List deserialize(String message) throws Throwable + { + return JsonUtils.JSON_OBJECT_MAPPER.readValue(message, TYPE_REFERENCE); + } } /** @@ -378,8 +537,8 @@ class MonitoringTask name(), numTimesReported, NANOSECONDS.toMillis(totalTimeNanos / numTimesReported), - NANOSECONDS.toMillis(minTime), - NANOSECONDS.toMillis(maxTime), + NANOSECONDS.toMillis(minTimeNanos), + NANOSECONDS.toMillis(maxTimeNanos), NANOSECONDS.toMillis(operation.timeoutNanos()), operation.isCrossNode() ? "msec/cross-node" : "msec"); } @@ -388,13 +547,21 @@ class MonitoringTask /** * An operation (query) that was reported as slow. */ - private final static class SlowOperation extends Operation + @VisibleForTesting + public final static class SlowOperation extends Operation { - SlowOperation(Monitorable operation, long failedAt) + // purely for deserialization purposes + public SlowOperation() + { + this(Monitorable.NO_OP, 0); + } + + public SlowOperation(Monitorable operation, long failedAt) { super(operation, failedAt); } + @JsonIgnore public String getLogMessage() { if (numTimesReported == 1) @@ -408,8 +575,8 @@ class MonitoringTask name(), numTimesReported, NANOSECONDS.toMillis(totalTimeNanos/ numTimesReported), - NANOSECONDS.toMillis(minTime), - NANOSECONDS.toMillis(maxTime), + NANOSECONDS.toMillis(minTimeNanos), + NANOSECONDS.toMillis(maxTimeNanos), NANOSECONDS.toMillis(operation.slowTimeoutNanos()), operation.isCrossNode() ? "msec/cross-node" : "msec"); } diff --git a/src/java/org/apache/cassandra/db/virtual/AbstractLoggerVirtualTable.java b/src/java/org/apache/cassandra/db/virtual/AbstractLoggerVirtualTable.java new file mode 100644 index 0000000000..008d5d432a --- /dev/null +++ b/src/java/org/apache/cassandra/db/virtual/AbstractLoggerVirtualTable.java @@ -0,0 +1,111 @@ +/* + * 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.virtual; + +import java.util.Collections; +import java.util.LinkedList; +import java.util.List; + +import com.google.common.annotations.VisibleForTesting; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import ch.qos.logback.classic.spi.LoggingEvent; +import org.apache.cassandra.schema.TableMetadata; + +/** + * This table is inherently limited on number of rows it can hold. + * + * @param type parameter saying what object is stored in internal bounded list for query purposes + */ +public abstract class AbstractLoggerVirtualTable extends AbstractMutableVirtualTable +{ + private static final Logger logger = LoggerFactory.getLogger(AbstractLoggerVirtualTable.class); + + // please be sure operations on this structure are thread-safe + protected final List buffer; + + @VisibleForTesting + protected static int resolveBufferSize(int wantedSize, int max, int defaultSize) + { + return (wantedSize < 1 || wantedSize > max) ? defaultSize : wantedSize; + } + + protected AbstractLoggerVirtualTable(TableMetadata metadata, int maxSize) + { + super(metadata); + this.buffer = BoundedLinkedList.create(maxSize); + logger.debug("capacity of virtual table {} is set to be at most {} rows", metadata().toString(), maxSize); + } + + public void add(LoggingEvent event) + { + List messages = getMessages(event); + if (messages != null) + { + // specifically calling buffer.add to reach BoundedLinkedList's add + // instead of linked list's addAll + for (U message : messages) + buffer.add(message); + } + } + + public abstract List getMessages(LoggingEvent event); + + @Override + public void truncate() + { + synchronized (buffer) + { + buffer.clear(); + } + } + + @Override + public boolean allowFilteringImplicitly() + { + return false; + } + + private static final class BoundedLinkedList extends LinkedList + { + private final int maxSize; + + public static List create(int size) + { + return Collections.synchronizedList(new BoundedLinkedList<>(size)); + } + + private BoundedLinkedList(int maxSize) + { + this.maxSize = maxSize; + } + + @Override + public synchronized boolean add(T t) + { + if (size() == maxSize) + removeLast(); + + addFirst(t); + + return true; + } + } +} diff --git a/src/java/org/apache/cassandra/db/virtual/LogMessagesTable.java b/src/java/org/apache/cassandra/db/virtual/LogMessagesTable.java index 5903ac2ab5..87978e3fd9 100644 --- a/src/java/org/apache/cassandra/db/virtual/LogMessagesTable.java +++ b/src/java/org/apache/cassandra/db/virtual/LogMessagesTable.java @@ -18,15 +18,11 @@ package org.apache.cassandra.db.virtual; -import java.util.Collections; import java.util.Date; import java.util.Iterator; -import java.util.LinkedList; import java.util.List; import com.google.common.annotations.VisibleForTesting; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import ch.qos.logback.classic.spi.LoggingEvent; import org.apache.cassandra.config.CassandraRelevantProperties; @@ -50,11 +46,8 @@ import org.apache.cassandra.schema.TableMetadata; * @see CASSANDRA-18238 * @see org.apache.cassandra.utils.logging.VirtualTableAppender */ -public final class LogMessagesTable extends AbstractMutableVirtualTable +public final class LogMessagesTable extends AbstractLoggerVirtualTable { - private static final Logger logger = LoggerFactory.getLogger(LogMessagesTable.class); - - public static final int LOGS_VIRTUAL_TABLE_MIN_ROWS = 1000; public static final int LOGS_VIRTUAL_TABLE_DEFAULT_ROWS = 50_000; public static final int LOGS_VIRTUAL_TABLE_MAX_ROWS = 100_000; @@ -67,11 +60,11 @@ public final class LogMessagesTable extends AbstractMutableVirtualTable public static final String LEVEL_COLUMN_NAME = "level"; public static final String MESSAGE_COLUMN_NAME = "message"; - private final List buffer; - LogMessagesTable(String keyspace) { - this(keyspace, resolveBufferSize()); + this(keyspace, resolveBufferSize(CassandraRelevantProperties.LOGS_VIRTUAL_TABLE_MAX_ROWS.getInt(), + LOGS_VIRTUAL_TABLE_MAX_ROWS, + LOGS_VIRTUAL_TABLE_DEFAULT_ROWS)); } @VisibleForTesting @@ -85,10 +78,14 @@ public final class LogMessagesTable extends AbstractMutableVirtualTable .addClusteringColumn(ORDER_IN_MILLISECOND_COLUMN_NAME, Int32Type.instance) .addRegularColumn(LOGGER_COLUMN_NAME, UTF8Type.instance) .addRegularColumn(LEVEL_COLUMN_NAME, UTF8Type.instance) - .addRegularColumn(MESSAGE_COLUMN_NAME, UTF8Type.instance).build()); + .addRegularColumn(MESSAGE_COLUMN_NAME, UTF8Type.instance).build(), + size); + } - logger.debug("capacity of virtual table {} is set to be at most {} rows", metadata().toString(), size); - buffer = BoundedLinkedList.create(size); + @Override + public List getMessages(LoggingEvent event) + { + return List.of(event); } @Override @@ -103,12 +100,12 @@ public final class LogMessagesTable extends AbstractMutableVirtualTable int index = 0; - Iterator iterator = buffer.listIterator(); + Iterator iterator = buffer.listIterator(); while (iterator.hasNext()) { - LogMessage log = iterator.next(); + LoggingEvent log = iterator.next(); - milliSecondsOfCurrentLog = log.timestamp; + milliSecondsOfCurrentLog = log.getTimeStamp(); if (milliSecondsOfPreviousLog == milliSecondsOfCurrentLog) ++index; else @@ -116,86 +113,13 @@ public final class LogMessagesTable extends AbstractMutableVirtualTable milliSecondsOfPreviousLog = milliSecondsOfCurrentLog; - result.row(new Date(log.timestamp), index) - .column(LOGGER_COLUMN_NAME, log.logger) - .column(LEVEL_COLUMN_NAME, log.level) - .column(MESSAGE_COLUMN_NAME, log.message); + result.row(new Date(milliSecondsOfCurrentLog), index) + .column(LOGGER_COLUMN_NAME, log.getLoggerName()) + .column(LEVEL_COLUMN_NAME, log.getLevel().toString()) + .column(MESSAGE_COLUMN_NAME, log.getFormattedMessage()); } } return result; } - - public void add(LoggingEvent event) - { - buffer.add(new LogMessage(event)); - } - - @Override - public void truncate() - { - buffer.clear(); - } - - @Override - public boolean allowFilteringImplicitly() - { - return false; - } - - @VisibleForTesting - static int resolveBufferSize() - { - int size = CassandraRelevantProperties.LOGS_VIRTUAL_TABLE_MAX_ROWS.getInt(); - return (size < LOGS_VIRTUAL_TABLE_MIN_ROWS || size > LOGS_VIRTUAL_TABLE_MAX_ROWS) - ? LOGS_VIRTUAL_TABLE_DEFAULT_ROWS : size; - } - - @VisibleForTesting - public static class LogMessage - { - public final long timestamp; - public final String logger; - public final String level; - public final String message; - - public LogMessage(LoggingEvent event) - { - this(event.getTimeStamp(), event.getLoggerName(), event.getLevel().toString(), event.getFormattedMessage()); - } - - public LogMessage(long timestamp, String logger, String level, String message) - { - this.timestamp = timestamp; - this.logger = logger; - this.level = level; - this.message = message; - } - } - - private static final class BoundedLinkedList extends LinkedList - { - private final int maxSize; - - public static List create(int size) - { - return Collections.synchronizedList(new BoundedLinkedList<>(size)); - } - - private BoundedLinkedList(int maxSize) - { - this.maxSize = maxSize; - } - - @Override - public boolean add(T t) - { - if (size() == maxSize) - removeLast(); - - addFirst(t); - - return true; - } - } } diff --git a/src/java/org/apache/cassandra/db/virtual/SlowQueriesTable.java b/src/java/org/apache/cassandra/db/virtual/SlowQueriesTable.java new file mode 100644 index 0000000000..0d392d0ce2 --- /dev/null +++ b/src/java/org/apache/cassandra/db/virtual/SlowQueriesTable.java @@ -0,0 +1,195 @@ +/* + * 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.virtual; + +import java.util.ArrayList; +import java.util.Date; +import java.util.Iterator; +import java.util.List; + +import com.google.common.annotations.VisibleForTesting; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import ch.qos.logback.classic.spi.LoggingEvent; +import org.apache.cassandra.config.CassandraRelevantProperties; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.marshal.BooleanType; +import org.apache.cassandra.db.marshal.Int32Type; +import org.apache.cassandra.db.marshal.LongType; +import org.apache.cassandra.db.marshal.TimestampType; +import org.apache.cassandra.db.marshal.UTF8Type; +import org.apache.cassandra.db.monitoring.MonitoringTask.Operation; +import org.apache.cassandra.dht.LocalPartitioner; +import org.apache.cassandra.schema.TableMetadata; + +import static java.util.concurrent.TimeUnit.NANOSECONDS; + +public class SlowQueriesTable extends AbstractLoggerVirtualTable +{ + private static final Logger logger = LoggerFactory.getLogger(SlowQueriesTable.class); + + public static final int LOGS_VIRTUAL_TABLE_DEFAULT_ROWS = 10_000; + public static final int LOGS_VIRTUAL_TABLE_MAX_ROWS = 100_000; + + public static final String TABLE_NAME = "slow_queries"; + private static final String TABLE_COMMENT = "Slow queries"; + + public static final String KEYSPACE_COLUMN_NAME = "keyspace_name"; + public static final String TABLE_COLUMN_NAME = "table_name"; + public static final String TIMESTAMP_COLUMN_NAME = "timestamp"; + public static final String QUERY_COLUMN_NAME = "query"; + public static final String MINIMUM_TIME_COLUMN_NAME = "min_ms"; + public static final String MAXIMUM_TIME_COLUMN_NAME = "max_ms"; + public static final String AVERAGE_TIME_COLUMN_NAME = "avg_ms"; + public static final String TIMES_REPORTED_COLUMN_NAME = "times_reported"; + public static final String CROSS_NODE_COLUMN_NAME = "cross_node"; + + SlowQueriesTable(String keyspace) + { + this(keyspace, resolveBufferSize(CassandraRelevantProperties.LOGS_SLOW_QUERIES_VIRTUAL_TABLE_MAX_ROWS.getInt(), + LOGS_VIRTUAL_TABLE_MAX_ROWS, + LOGS_VIRTUAL_TABLE_DEFAULT_ROWS)); + } + + @VisibleForTesting + SlowQueriesTable(String keyspace, int size) + { + super(TableMetadata.builder(keyspace, TABLE_NAME) + .comment(TABLE_COMMENT) + .kind(TableMetadata.Kind.VIRTUAL) + .partitioner(new LocalPartitioner(UTF8Type.instance)) + .addPartitionKeyColumn(KEYSPACE_COLUMN_NAME, UTF8Type.instance) + .addClusteringColumn(TABLE_COLUMN_NAME, UTF8Type.instance) + .addClusteringColumn(TIMESTAMP_COLUMN_NAME, TimestampType.instance) + // We are adding query as a clustering column for uniqueness, + // In theory, it might happen that two monitoring operations + // would be emitted for same keyspace, same table at the exact same time + // (in milliseconds). That means that one operation would "shadow" + // another one because primary key would be same for both. + // To make it truly unique, we include query among clustering keys + // as well. If queries were same, then they would be also reported so + // (it would be reflected in "times_reported" column) + .addClusteringColumn(QUERY_COLUMN_NAME, UTF8Type.instance) + .addRegularColumn(MINIMUM_TIME_COLUMN_NAME, LongType.instance) + .addRegularColumn(MAXIMUM_TIME_COLUMN_NAME, LongType.instance) + .addRegularColumn(AVERAGE_TIME_COLUMN_NAME, LongType.instance) + .addRegularColumn(TIMES_REPORTED_COLUMN_NAME, Int32Type.instance) + .addRegularColumn(CROSS_NODE_COLUMN_NAME, BooleanType.instance) + .build(), + size); + } + + @Override + protected void applyPartitionDeletion(ColumnValues partitionKey) + { + String keyspace = partitionKey.value(0); + + synchronized (buffer) + { + buffer.removeIf(o -> o.keyspace().equals(keyspace)); + } + } + + @Override + public DataSet data() + { + SimpleDataSet result = new SimpleDataSet(metadata(), DecoratedKey.comparator.reversed()); + + synchronized (buffer) + { + Iterator iterator = buffer.listIterator(); + while (iterator.hasNext()) + { + Operation operation = iterator.next(); + + result.row(operation.keyspace(), operation.table(), new Date(operation.timestampMs()), operation.name()) + .column(MINIMUM_TIME_COLUMN_NAME, NANOSECONDS.toMillis(operation.minTimeNanos())) + .column(MAXIMUM_TIME_COLUMN_NAME, NANOSECONDS.toMillis(operation.maxTimeNanos())) + .column(AVERAGE_TIME_COLUMN_NAME, NANOSECONDS.toMillis(operation.averageTime())) + .column(TIMES_REPORTED_COLUMN_NAME, operation.numTimesReported()) + .column(CROSS_NODE_COLUMN_NAME, operation.isCrossNode()); + } + } + + return result; + } + + @Override + public List getMessages(LoggingEvent event) + { + try + { + List qualified = new ArrayList<>(); + for (Operation operation : Operation.deserialize(event.getMessage())) + { + + // in (improbable) case there is an operation which does not have + // keyspace / table on it, we just skip this from processing + // as we would have nothing to show for partition key and clustering column + if (operation.keyspace() == null || operation.table() == null) + continue; + + // if cf of an operation is present, take keyspace and table name from it + // instead of having new string instances per operation which might + // take relatively a lot of additional space unnecessarily + Keyspace keyspace = Keyspace.openIfExists(operation.keyspace()); + String keyspaceName; + String tableName; + if (keyspace != null) + { + keyspaceName = keyspace.getName(); + try + { + ColumnFamilyStore table = keyspace.getColumnFamilyStore(operation.table()); + tableName = table.getTableName(); + } + catch (IllegalArgumentException ex) + { + tableName = operation.table(); + } + } + else + { + keyspaceName = operation.keyspace(); + tableName = operation.table(); + } + + operation.setKeyspace(keyspaceName); + operation.setTable(tableName); + qualified.add(operation); + } + + return qualified; + } + catch (Throwable t) + { + logger.trace("Unable to generate list of slow queries", t); + return null; + } + } + + @Override + public boolean allowFilteringImplicitly() + { + return true; + } +} diff --git a/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java b/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java index 3ca8f728a8..28c6dc8fef 100644 --- a/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java +++ b/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java @@ -61,6 +61,7 @@ public final class SystemViewsKeyspace extends VirtualKeyspace .add(new GossipInfoTable(VIRTUAL_VIEWS)) .add(new QueriesTable(VIRTUAL_VIEWS)) .add(new LogMessagesTable(VIRTUAL_VIEWS)) + .add(new SlowQueriesTable(VIRTUAL_VIEWS)) .add(new SnapshotsTable(VIRTUAL_VIEWS)) .add(new PeersTable(VIRTUAL_VIEWS)) .add(new LocalTable(VIRTUAL_VIEWS)) diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index d3c787d2e2..171ec47e1e 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -59,6 +59,8 @@ import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.db.SystemKeyspaceMigrator41; import org.apache.cassandra.db.commitlog.CommitLog; import org.apache.cassandra.db.virtual.AccordDebugKeyspace; +import org.apache.cassandra.db.virtual.LogMessagesTable; +import org.apache.cassandra.db.virtual.SlowQueriesTable; import org.apache.cassandra.db.virtual.SystemViewsKeyspace; import org.apache.cassandra.db.virtual.VirtualKeyspace; import org.apache.cassandra.db.virtual.VirtualKeyspaceRegistry; @@ -94,6 +96,7 @@ import org.apache.cassandra.utils.NativeLibrary; import org.apache.cassandra.utils.concurrent.Future; import org.apache.cassandra.utils.concurrent.FutureCombiner; import org.apache.cassandra.utils.logging.LoggingSupportFactory; +import org.apache.cassandra.utils.logging.SlowQueriesAppender; import org.apache.cassandra.utils.logging.VirtualTableAppender; import static java.util.concurrent.TimeUnit.NANOSECONDS; @@ -436,7 +439,6 @@ public class CassandraDaemon { exitOrFail(e.returnCode, e.getMessage(), e.getCause()); } - } /** @@ -555,11 +557,17 @@ public class CassandraDaemon if (DatabaseDescriptor.getAccord().enable_virtual_debug_only_keyspace) VirtualKeyspaceRegistry.instance.register(AccordDebugKeyspace.instance); - // flush log messages to system_views.system_logs virtual table as there were messages already logged - // before that virtual table was instantiated + // Flush log messages to system_views.system_logs virtual table as there were messages already logged + // before that virtual table was instantiated. + // In general, there is no need to do same treatment for slow queries as by the time queries are processed + // the logging framework if fully setup already but for the sake of it and to be sure, just do it as well. LoggingSupportFactory.getLoggingSupport() .getAppender(VirtualTableAppender.class, VirtualTableAppender.APPENDER_NAME) - .ifPresent(appender -> ((VirtualTableAppender) appender).flushBuffer()); + .ifPresent(appender -> appender.flushBuffer(LogMessagesTable.class, LogMessagesTable.TABLE_NAME)); + + LoggingSupportFactory.getLoggingSupport() + .getAppender(SlowQueriesAppender.class, SlowQueriesAppender.APPENDER_NAME) + .ifPresent(appender -> appender.flushBuffer(SlowQueriesTable.class, SlowQueriesTable.TABLE_NAME)); } public synchronized void initializeClientTransports() diff --git a/src/java/org/apache/cassandra/utils/logging/AbstractVirtualTableAppender.java b/src/java/org/apache/cassandra/utils/logging/AbstractVirtualTableAppender.java new file mode 100644 index 0000000000..7becbc13fc --- /dev/null +++ b/src/java/org/apache/cassandra/utils/logging/AbstractVirtualTableAppender.java @@ -0,0 +1,144 @@ +/* + * 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.utils.logging; + +import java.util.LinkedList; +import java.util.List; +import java.util.Optional; + +import ch.qos.logback.classic.spi.LoggingEvent; +import ch.qos.logback.core.AppenderBase; +import org.apache.cassandra.db.virtual.AbstractLoggerVirtualTable; +import org.apache.cassandra.db.virtual.SlowQueriesTable; +import org.apache.cassandra.db.virtual.VirtualKeyspace; +import org.apache.cassandra.db.virtual.VirtualKeyspaceRegistry; +import org.apache.cassandra.db.virtual.VirtualTable; + +import static org.apache.cassandra.schema.SchemaConstants.VIRTUAL_VIEWS; + +public abstract class AbstractVirtualTableAppender extends AppenderBase +{ + private final int defaultRows; + + protected AbstractVirtualTableAppender(int defaultRows) + { + this.defaultRows = defaultRows; + } + + // for holding messages until virtual registry contains logs virtual table + // as it takes some time during startup of a node to initialise virtual tables but messages are + // logged already + protected final List messageBuffer = new LinkedList<>(); + + protected T getVirtualTable(Class vtableClass, String tableName) + { + VirtualKeyspace keyspace = VirtualKeyspaceRegistry.instance.getKeyspaceNullable(VIRTUAL_VIEWS); + + if (keyspace == null) + return null; + + Optional virtualTable = keyspace.tables() + .stream() + .filter(vt -> vt.name().equals(tableName)) + .findFirst(); + + if (virtualTable.isEmpty()) + return null; + + VirtualTable vt = virtualTable.get(); + + if (!vt.getClass().equals(vtableClass)) + throw new IllegalStateException(String.format("Virtual table %s.%s is not backed by an instance of %s but by %s", + VIRTUAL_VIEWS, + tableName, + vtableClass.getName(), + vt.getClass().getName())); + + return (T) vt; + } + + /** + * This method adds an event to virtual table, when present. + * When vtable is null, we will attempt to find it among registered ones. Then not found, we add it to internal + * buffer for later processing. This might happen e.g. for logging tables when log events + * were appended via logging framework sooner than registration of virtual tables was done so after they are registered, + * they would miss logging events happened before being so. + * + * @param vtable vtable to append to + * @param event event to append to + * @param tableName table name of virtual table to append to + * @return vtable or when null, found vtable + */ + protected AbstractLoggerVirtualTable appendToVirtualTable(AbstractLoggerVirtualTable vtable, LoggingEvent event, String tableName) + { + AbstractLoggerVirtualTable foundVtable; + if (vtable == null) + { + foundVtable = getVirtualTable(SlowQueriesTable.class, tableName); + if (foundVtable == null) + addToBuffer(event); + else + foundVtable.add(event); + } + else + { + foundVtable = vtable; + vtable.add(event); + } + + return foundVtable; + } + + @Override + public void stop() + { + synchronized (messageBuffer) + { + messageBuffer.clear(); + super.stop(); + } + } + + /** + * Flushes all log entries which were appended before virtual table was registered. + * + * @see org.apache.cassandra.service.CassandraDaemon#setupVirtualKeyspaces + */ + public void flushBuffer(Class> vtableClass, String tableName) + { + synchronized (messageBuffer) + { + Optional.ofNullable(getVirtualTable(vtableClass, tableName)).ifPresent(vtable -> { + messageBuffer.forEach(vtable::add); + messageBuffer.clear(); + }); + } + } + + protected void addToBuffer(LoggingEvent eventObject) + { + synchronized (messageBuffer) + { + // we restrict how many logging events we can put into buffer, + // so we are not growing without any bound when things go south + if (messageBuffer.size() < defaultRows) + messageBuffer.add(eventObject); + } + } +} diff --git a/src/java/org/apache/cassandra/utils/logging/LogbackLoggingSupport.java b/src/java/org/apache/cassandra/utils/logging/LogbackLoggingSupport.java index e710d44dd1..f32963b73a 100644 --- a/src/java/org/apache/cassandra/utils/logging/LogbackLoggingSupport.java +++ b/src/java/org/apache/cassandra/utils/logging/LogbackLoggingSupport.java @@ -60,7 +60,8 @@ public class LogbackLoggingSupport implements LoggingSupport @Override public void onStartup() { - checkOnlyOneVirtualTableAppender(); + checkOnlyOneVirtualTableAppender(VirtualTableAppender.class); + checkOnlyOneVirtualTableAppender(SlowQueriesAppender.class); // The default logback configuration in conf/logback.xml allows reloading the // configuration when the configuration file has changed (every 60 seconds by default). @@ -138,7 +139,20 @@ public class LogbackLoggingSupport implements LoggingSupport } @Override - public Optional> getAppender(Class appenderClass, String name) + public Optional getLogger(String loggerName) + { + LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory(); + for (Logger logBackLogger : lc.getLoggerList()) + { + if (logBackLogger.getName().equals(loggerName)) + return Optional.of(logBackLogger); + } + + return Optional.empty(); + } + + @Override + public > Optional getAppender(Class appenderClass, String appenderName) { LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory(); for (Logger logBackLogger : lc.getLoggerList()) @@ -146,15 +160,15 @@ public class LogbackLoggingSupport implements LoggingSupport for (Iterator> iterator = logBackLogger.iteratorForAppenders(); iterator.hasNext();) { Appender appender = iterator.next(); - if (appender.getClass() == appenderClass && appender.getName().equals(name)) - return Optional.of(appender); + if (appender.getClass() == appenderClass && appender.getName().equals(appenderName)) + return Optional.of(appenderClass.cast(appender)); } } return Optional.empty(); } - private void checkOnlyOneVirtualTableAppender() + private void checkOnlyOneVirtualTableAppender(Class appenderClass) { int count = 0; LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory(); @@ -164,7 +178,7 @@ public class LogbackLoggingSupport implements LoggingSupport for (Iterator> iterator = logBackLogger.iteratorForAppenders(); iterator.hasNext();) { Appender appender = iterator.next(); - if (appender instanceof VirtualTableAppender) + if (appenderClass.isAssignableFrom(appender.getClass())) { virtualAppenderNames.add(appender.getName()); count += 1; @@ -174,7 +188,7 @@ public class LogbackLoggingSupport implements LoggingSupport if (count > 1) throw new IllegalStateException(String.format("There are multiple appenders of class %s of names %s. There is only one appender of such class allowed.", - VirtualTableAppender.class.getName(), String.join(",", virtualAppenderNames))); + appenderClass.getName(), String.join(",", virtualAppenderNames))); } private boolean hasAppenders(Logger logBackLogger) diff --git a/src/java/org/apache/cassandra/utils/logging/LoggingSupport.java b/src/java/org/apache/cassandra/utils/logging/LoggingSupport.java index 35e11975f9..00b40cb966 100644 --- a/src/java/org/apache/cassandra/utils/logging/LoggingSupport.java +++ b/src/java/org/apache/cassandra/utils/logging/LoggingSupport.java @@ -21,6 +21,7 @@ package org.apache.cassandra.utils.logging; import java.util.Map; import java.util.Optional; +import ch.qos.logback.classic.Logger; import ch.qos.logback.core.Appender; /** @@ -53,7 +54,12 @@ public interface LoggingSupport */ Map getLoggingLevels(); - default Optional> getAppender(Class appenderClass, String appenderName) + default > Optional getAppender(Class appenderClass, String appenderName) + { + return Optional.empty(); + } + + default Optional getLogger(String loggerName) { return Optional.empty(); } diff --git a/src/java/org/apache/cassandra/utils/logging/SlowQueriesAppender.java b/src/java/org/apache/cassandra/utils/logging/SlowQueriesAppender.java new file mode 100644 index 0000000000..4af2e38307 --- /dev/null +++ b/src/java/org/apache/cassandra/utils/logging/SlowQueriesAppender.java @@ -0,0 +1,45 @@ +/* + * 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.utils.logging; + +import ch.qos.logback.classic.spi.LoggingEvent; +import org.apache.cassandra.db.virtual.AbstractLoggerVirtualTable; +import org.apache.cassandra.db.virtual.SlowQueriesTable; + +public final class SlowQueriesAppender extends AbstractVirtualTableAppender +{ + public static final String APPENDER_NAME = "SLOW_QUERIES_APPENDER"; + + private AbstractLoggerVirtualTable slowQueries; + + public SlowQueriesAppender() + { + super(SlowQueriesTable.LOGS_VIRTUAL_TABLE_DEFAULT_ROWS); + } + + @Override + protected void append(LoggingEvent eventObject) + { + // slowQueries will be null as long as virtual tables + // are not registered, and we already try to put queries there. + // As soon as vtable is registered (as part of node's startup / initialisation), + // slow queries will never be null again + slowQueries = appendToVirtualTable(slowQueries, eventObject, SlowQueriesTable.TABLE_NAME); + } +} diff --git a/src/java/org/apache/cassandra/utils/logging/VirtualTableAppender.java b/src/java/org/apache/cassandra/utils/logging/VirtualTableAppender.java index 2820b2936f..03a142004a 100644 --- a/src/java/org/apache/cassandra/utils/logging/VirtualTableAppender.java +++ b/src/java/org/apache/cassandra/utils/logging/VirtualTableAppender.java @@ -18,111 +18,35 @@ package org.apache.cassandra.utils.logging; -import java.util.LinkedList; -import java.util.List; -import java.util.Optional; import java.util.Set; import com.google.common.collect.ImmutableSet; import ch.qos.logback.classic.spi.LoggingEvent; -import ch.qos.logback.core.AppenderBase; import org.apache.cassandra.audit.FileAuditLogger; +import org.apache.cassandra.db.virtual.AbstractLoggerVirtualTable; import org.apache.cassandra.db.virtual.LogMessagesTable; -import org.apache.cassandra.db.virtual.VirtualKeyspace; -import org.apache.cassandra.db.virtual.VirtualKeyspaceRegistry; -import org.apache.cassandra.db.virtual.VirtualTable; - -import static org.apache.cassandra.db.virtual.LogMessagesTable.LOGS_VIRTUAL_TABLE_DEFAULT_ROWS; -import static org.apache.cassandra.db.virtual.LogMessagesTable.TABLE_NAME; -import static org.apache.cassandra.schema.SchemaConstants.VIRTUAL_VIEWS; /** * Appends Cassandra logs to virtual table system_views.system_logs */ -public final class VirtualTableAppender extends AppenderBase +public final class VirtualTableAppender extends AbstractVirtualTableAppender { public static final String APPENDER_NAME = "CQLLOG"; private static final Set forbiddenLoggers = ImmutableSet.of(FileAuditLogger.class.getName()); - private LogMessagesTable logs; + private AbstractLoggerVirtualTable logs; - // for holding messages until virtual registry contains logs virtual table - // as it takes some time during startup of a node to initialise virtual tables but messages are - // logged already - private final List messageBuffer = new LinkedList<>(); + public VirtualTableAppender() + { + super(LogMessagesTable.LOGS_VIRTUAL_TABLE_DEFAULT_ROWS); + } @Override protected void append(LoggingEvent eventObject) { if (!forbiddenLoggers.contains(eventObject.getLoggerName())) - { - if (logs == null) - { - logs = getVirtualTable(); - if (logs == null) - addToBuffer(eventObject); - else - logs.add(eventObject); - } - else - logs.add(eventObject); - } - } - - @Override - public void stop() - { - messageBuffer.clear(); - super.stop(); - } - - /** - * Flushes all logs which were appended before virtual table was registered. - * - * @see org.apache.cassandra.service.CassandraDaemon#setupVirtualKeyspaces - */ - public void flushBuffer() - { - Optional.ofNullable(getVirtualTable()).ifPresent(vtable -> { - messageBuffer.forEach(vtable::add); - messageBuffer.clear(); - }); - } - - private LogMessagesTable getVirtualTable() - { - VirtualKeyspace keyspace = VirtualKeyspaceRegistry.instance.getKeyspaceNullable(VIRTUAL_VIEWS); - - if (keyspace == null) - return null; - - Optional logsTable = keyspace.tables() - .stream() - .filter(vt -> vt.name().equals(TABLE_NAME)) - .findFirst(); - - if (!logsTable.isPresent()) - return null; - - VirtualTable vt = logsTable.get(); - - if (!(vt instanceof LogMessagesTable)) - throw new IllegalStateException(String.format("Virtual table %s.%s is not backed by an instance of %s but by %s", - VIRTUAL_VIEWS, - TABLE_NAME, - LogMessagesTable.class.getName(), - vt.getClass().getName())); - - return (LogMessagesTable) vt; - } - - private void addToBuffer(LoggingEvent eventObject) - { - // we restrict how many logging events we can put into buffer, - // so we are not growing without any bound when things go south - if (messageBuffer.size() < LOGS_VIRTUAL_TABLE_DEFAULT_ROWS) - messageBuffer.add(eventObject); + logs = appendToVirtualTable(logs, eventObject, LogMessagesTable.TABLE_NAME); } } diff --git a/test/conf/logback-dtest_with_slow_query_appender.xml b/test/conf/logback-dtest_with_slow_query_appender.xml new file mode 100644 index 0000000000..1b6ed7511f --- /dev/null +++ b/test/conf/logback-dtest_with_slow_query_appender.xml @@ -0,0 +1,63 @@ + + + + + + + + + + + ./build/test/logs/${cassandra.testtag}/${suitename}/${cluster_id}/${instance_id}/system.log + + %-5level [%thread] ${instance_id} %date{ISO8601} %msg%n + + true + + + + + %-5level %date{HH:mm:ss,SSS} %msg%n + + + WARN + + + + + + %-5level [%thread] ${instance_id} %date{ISO8601} %F:%L - %msg%n + + + DEBUG + + + + + + + + + + + + + + + diff --git a/test/conf/logback-dtest_with_slow_query_appender_invalid.xml b/test/conf/logback-dtest_with_slow_query_appender_invalid.xml new file mode 100644 index 0000000000..a2252dd23a --- /dev/null +++ b/test/conf/logback-dtest_with_slow_query_appender_invalid.xml @@ -0,0 +1,73 @@ + + + + + + + + + + + ./build/test/logs/${cassandra.testtag}/${suitename}/${cluster_id}/${instance_id}/system.log + + %-5level [%thread] ${instance_id} %date{ISO8601} %msg%n + + true + + + + + %-5level %date{HH:mm:ss,SSS} %msg%n + + + WARN + + + + + + %-5level [%thread] ${instance_id} %date{ISO8601} %F:%L - %msg%n + + + DEBUG + + + + + + + + + INFO + + + + + + + + + + + + + + + + diff --git a/test/distributed/org/apache/cassandra/distributed/test/AbstractVirtualLogsTableTest.java b/test/distributed/org/apache/cassandra/distributed/test/AbstractVirtualLogsTableTest.java new file mode 100644 index 0000000000..c8bb32fe72 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/AbstractVirtualLogsTableTest.java @@ -0,0 +1,34 @@ +/* + * 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; + +import org.junit.Ignore; + +import static java.lang.String.format; + +@Ignore +public abstract class AbstractVirtualLogsTableTest extends TestBaseImpl +{ + public String query(String template) + { + return format(template, getTableName()); + } + + public abstract String getTableName(); +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/SlowQueriesAppenderTest.java b/test/distributed/org/apache/cassandra/distributed/test/SlowQueriesAppenderTest.java new file mode 100644 index 0000000000..d8b6b3f00f --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/SlowQueriesAppenderTest.java @@ -0,0 +1,73 @@ +/* + * 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; + +import org.junit.Test; + +import org.apache.cassandra.db.virtual.SlowQueriesTable; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.Constants; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.shared.WithProperties; +import org.apache.cassandra.schema.SchemaConstants; +import org.apache.cassandra.utils.logging.SlowQueriesAppender; + +import static java.lang.String.format; +import static org.apache.cassandra.config.CassandraRelevantProperties.LOGBACK_CONFIGURATION_FILE; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +/** + * It is inherently tricky / flaky to make some queries to be slow so we just test + * the invalid configuration otherwise the table as such is tested in {@link org.apache.cassandra.db.virtual.SlowQueriesTableTest}. + */ +public class SlowQueriesAppenderTest extends AbstractVirtualLogsTableTest +{ + @Test + public void testMultipleAppendersFailToStartNode() throws Throwable + { + LOGBACK_CONFIGURATION_FILE.setString("test/conf/logback-dtest_with_slow_query_appender_invalid.xml"); + + // NOTE: Because cluster startup is expected to fail in this case, and can leave things in a weird state + // for the next state, create without starting, and set failure as shutdown to false, + // so the try-with-resources can close instances properly. + try (WithProperties properties = new WithProperties().set(LOGBACK_CONFIGURATION_FILE, "test/conf/logback-dtest_with_slow_query_appender_invalid.xml"); + Cluster cluster = Cluster.build(1) + .withConfig(c -> c.with(Feature.values()) + .set(Constants.KEY_DTEST_API_STARTUP_FAILURE_AS_SHUTDOWN, false)) + .createWithoutStarting()) + { + cluster.startup(); + fail("Node should not start as there is supposed to be invalid logback configuration file."); + } + catch (IllegalStateException ex) + { + assertEquals(format("There are multiple appenders of class %s " + + "of names SLOW_QUERIES_APPENDER,SLOW_QUERIES_APPENDER_2. There is only one appender of such class allowed.", + SlowQueriesAppender.class.getName()), + ex.getMessage()); + } + } + + @Override + public String getTableName() + { + return format("%s.%s", SchemaConstants.VIRTUAL_VIEWS, SlowQueriesTable.TABLE_NAME); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/SlowQueryDeserTest.java b/test/distributed/org/apache/cassandra/distributed/test/SlowQueryDeserTest.java new file mode 100644 index 0000000000..ae5bcc966c --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/SlowQueryDeserTest.java @@ -0,0 +1,66 @@ +/* + * 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; + +import java.util.Collection; +import java.util.List; + +import org.junit.Test; + +import org.apache.cassandra.db.monitoring.MonitorableImpl; +import org.apache.cassandra.db.monitoring.MonitoringTask; +import org.apache.cassandra.db.monitoring.MonitoringTask.SlowOperation; +import org.apache.cassandra.utils.Clock; + +public class SlowQueryDeserTest +{ + @Test + public void testSlowQueryDeser() throws Throwable + { + SlowOperation slowOperation = new SlowOperation(new MonitorableImpl() + { + @Override + public String name() + { + return String.format("select * from %s.%s where id = 5", monitoredOnKeyspace(), monitoredOnTable()); + } + + @Override + public String monitoredOnKeyspace() + { + return "ks"; + } + + @Override + public String monitoredOnTable() + { + return "tb"; + } + + @Override + public boolean isCrossNode() + { + return true; + } + }, Clock.Global.currentTimeMillis()); + + String serialize = MonitoringTask.Operation.serialize(List.of(slowOperation)); + Collection deserialize = MonitoringTask.Operation.deserialize(serialize); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/VirtualTableLogsTest.java b/test/distributed/org/apache/cassandra/distributed/test/VirtualTableLogsTest.java index 71ef4dbe78..bf9f58123e 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/VirtualTableLogsTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/VirtualTableLogsTest.java @@ -25,7 +25,6 @@ import org.junit.Test; import ch.qos.logback.classic.Level; import org.apache.cassandra.db.virtual.LogMessagesTable; -import org.apache.cassandra.db.virtual.LogMessagesTable.LogMessage; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.Constants; import org.apache.cassandra.distributed.api.Feature; @@ -47,8 +46,14 @@ import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; -public class VirtualTableLogsTest extends TestBaseImpl +public class VirtualTableLogsTest extends AbstractVirtualLogsTableTest { + @Override + public String getTableName() + { + return format("%s.%s", SchemaConstants.VIRTUAL_VIEWS, LogMessagesTable.TABLE_NAME); + } + @Test public void testVTableOutput() throws Throwable { @@ -56,9 +61,9 @@ public class VirtualTableLogsTest extends TestBaseImpl Cluster cluster = Cluster.build(1) .withConfig(c -> c.with(Feature.values())) .start(); - ) + ) { - List rows = getRows(cluster); + List rows = getRows(cluster); assertFalse(rows.isEmpty()); rows.forEach(message -> assertTrue(Level.toLevel(message.level).isGreaterOrEqual(Level.INFO))); @@ -91,39 +96,36 @@ public class VirtualTableLogsTest extends TestBaseImpl } } - private List getRows(Cluster cluster) + private List getRows(Cluster cluster) { SimpleQueryResult simpleQueryResult = cluster.coordinator(1).executeWithResult(query("select * from %s"), ONE); - List rows = new ArrayList<>(); + List rows = new ArrayList<>(); simpleQueryResult.forEachRemaining(row -> { long timestamp = row.getTimestamp(TIMESTAMP_COLUMN_NAME).getTime(); String logger = row.getString(LOGGER_COLUMN_NAME); String level = row.getString(LEVEL_COLUMN_NAME); String message = row.getString(MESSAGE_COLUMN_NAME); int order = row.getInteger(ORDER_IN_MILLISECOND_COLUMN_NAME); - TestingLogMessage logMessage = new TestingLogMessage(timestamp, logger, level, message, order); + LogMessage logMessage = new LogMessage(timestamp, logger, level, message, order); rows.add(logMessage); }); return rows; } - private String query(String template) + private static class LogMessage { - return format(template, getTableName()); - } + public final long timestamp; + public final String logger; + public final String level; + public final String message; + public final int order; - private String getTableName() - { - return format("%s.%s", SchemaConstants.VIRTUAL_VIEWS, LogMessagesTable.TABLE_NAME); - } - - private static class TestingLogMessage extends LogMessage - { - private int order; - - public TestingLogMessage(long timestamp, String logger, String level, String message, int order) + public LogMessage(long timestamp, String logger, String level, String message, int order) { - super(timestamp, logger, level, message); + this.timestamp = timestamp; + this.logger = logger; + this.level = level; + this.message = message; this.order = order; } } diff --git a/test/unit/org/apache/cassandra/db/virtual/AbstractLoggerVirtualTableTest.java b/test/unit/org/apache/cassandra/db/virtual/AbstractLoggerVirtualTableTest.java new file mode 100644 index 0000000000..40a926fda7 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/virtual/AbstractLoggerVirtualTableTest.java @@ -0,0 +1,169 @@ +/* + * 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.virtual; + +import java.time.Instant; +import java.util.Date; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.List; + +import com.google.common.collect.ImmutableList; +import org.junit.Ignore; +import org.junit.Test; + +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.spi.LoggingEvent; +import com.datastax.driver.core.Row; +import org.apache.cassandra.cql3.CQLTester; +import org.apache.cassandra.db.DataRange; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +@Ignore +public abstract class AbstractLoggerVirtualTableTest extends CQLTester +{ + protected final String keyspace = createKeyspaceName(); + + protected AbstractLoggerVirtualTable table; + + @Test + public void testTruncate() + { + registerTable(); + + int numberOfRows = 100; + List loggingEvents = getLoggingEvents(numberOfRows); + loggingEvents.forEach(table::add); + + execute(query("truncate %s")); + + assertTrue(executeNet(query("select timestamp from %s")).all().isEmpty()); + } + + @Test + public void testEmpty() throws Throwable + { + registerTable(); + assertEmpty(execute(query("select * from %s"))); + } + + @Test + public void testInsert() + { + registerTable(); + + int numberOfRows = 1000; + List loggingEvents = getLoggingEvents(numberOfRows); + loggingEvents.forEach(table::add); + + assertEquals(numberOfRows, execute(query("select * from %s")).size()); + } + + @Test + public void testLimitedCapacity() + { + registerTable(100); + + int numberOfRows = 1000; + List loggingEvents = getLoggingEvents(numberOfRows); + loggingEvents.forEach(table::add); + + // even we inserted 1000 rows, only 100 are present as its capacity is bounded + assertEquals(100, numberOfPartitions()); + + // the first record in the table will be the last one which we inserted + LoggingEvent firstEvent = loggingEvents.get(999); + assertRowsNet(executeNet(query("select timestamp from %s limit 1")), + new Object[]{ new Date(firstEvent.getTimeStamp()) }); + + // the last record in the table will be 900th we inserted + List all = executeNet(query("select timestamp from %s")).all(); + assertEquals(100, all.size()); + Row row = all.get(all.size() - 1); + Date timestamp = row.getTimestamp(0); + assertEquals(loggingEvents.get(900).getTimeStamp(), timestamp.getTime()); + } + + protected abstract void registerTable(int maxSize); + + protected abstract void registerTable(); + + protected void registerVirtualTable(AbstractLoggerVirtualTable table) + { + this.table = table; + VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(table.metadata.keyspace, ImmutableList.of(this.table))); + } + + protected String query(String query) + { + return String.format(query, table.toString()); + } + + protected List getLoggingEvents(int size) + { + return getLoggingEvents(size, Instant.now(), 1); + } + + protected List getLoggingEvents(int size, Instant firstTimestamp, int logsInMillisecond) + { + List logs = new LinkedList<>(); + int partitions = size / logsInMillisecond; + + for (int i = 0; i < partitions; i++) + { + firstTimestamp = firstTimestamp.plusSeconds(i); + + for (int j = 0; j < logsInMillisecond; j++) + logs.add(getLoggingEvent(firstTimestamp.toEpochMilli())); + } + + return logs; + } + + protected int numberOfPartitions() + { + AbstractVirtualTable.DataSet data = table.data(); + Iterator partitions = data.getPartitions(DataRange.allData(table.metadata.partitioner)); + int numberOfPartitions = 0; + + while (partitions.hasNext()) + { + partitions.next(); + numberOfPartitions += 1; + } + + return numberOfPartitions; + } + + protected LoggingEvent getLoggingEvent(long timestamp) + { + LoggingEvent event = new LoggingEvent(); + event.setLevel(Level.INFO); + event.setMessage(getMessage(timestamp)); + event.setLoggerName(AbstractLoggerVirtualTableTest.class.getName()); + event.setThreadName(Thread.currentThread().getName()); + event.setTimeStamp(timestamp); + + return event; + } + + protected abstract String getMessage(long timestamp); +} diff --git a/test/unit/org/apache/cassandra/db/virtual/LogMessagesTableTest.java b/test/unit/org/apache/cassandra/db/virtual/LogMessagesTableTest.java index dd32058533..7025e8ad4c 100644 --- a/test/unit/org/apache/cassandra/db/virtual/LogMessagesTableTest.java +++ b/test/unit/org/apache/cassandra/db/virtual/LogMessagesTableTest.java @@ -20,69 +20,34 @@ package org.apache.cassandra.db.virtual; import java.time.Instant; import java.util.Date; -import java.util.Iterator; -import java.util.LinkedList; import java.util.List; -import com.google.common.collect.ImmutableList; import org.junit.Test; -import ch.qos.logback.classic.Level; import ch.qos.logback.classic.spi.LoggingEvent; import com.datastax.driver.core.Row; -import org.apache.cassandra.cql3.CQLTester; -import org.apache.cassandra.db.DataRange; -import org.apache.cassandra.db.marshal.TimestampType; -import org.apache.cassandra.db.virtual.AbstractVirtualTable.DataSet; -import org.apache.cassandra.db.virtual.AbstractVirtualTable.Partition; -import org.apache.cassandra.dht.LocalPartitioner; import static org.apache.cassandra.config.CassandraRelevantProperties.LOGS_VIRTUAL_TABLE_MAX_ROWS; +import static org.apache.cassandra.db.virtual.LogMessagesTable.LOGS_VIRTUAL_TABLE_DEFAULT_ROWS; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; -public class LogMessagesTableTest extends CQLTester +public class LogMessagesTableTest extends AbstractLoggerVirtualTableTest { - private String keyspace = createKeyspaceName(); - private LogMessagesTable table; - @Test - public void testTruncate() throws Throwable + public void testMultipleLogsInSameMillisecond() { - registerVirtualTable(); - - int numberOfRows = 100; - List loggingEvents = getLoggingEvents(numberOfRows); + registerTable(); + List loggingEvents = getLoggingEvents(10, Instant.now(), 5); loggingEvents.forEach(table::add); - execute(query("truncate %s")); - - assertTrue(executeNet(query("select timestamp from %s")).all().isEmpty()); + // 2 partitions, 5 rows in each + assertEquals(2, numberOfPartitions()); } @Test - public void empty() throws Throwable + public void testLimitedCapacity() { - registerVirtualTable(); - assertEmpty(execute(query("select * from %s"))); - } - - @Test - public void testInsert() - { - registerVirtualTable(); - - int numberOfRows = 1000; - List loggingEvents = getLoggingEvents(numberOfRows); - loggingEvents.forEach(table::add); - - assertEquals(numberOfRows, numberOfPartitions()); - } - - @Test - public void testLimitedCapacity() throws Throwable - { - registerVirtualTable(100); + registerTable(100); int numberOfRows = 1000; List loggingEvents = getLoggingEvents(numberOfRows); @@ -94,7 +59,7 @@ public class LogMessagesTableTest extends CQLTester // the first record in the table will be the last one which we inserted LoggingEvent firstEvent = loggingEvents.get(999); assertRowsNet(executeNet(query("select timestamp from %s limit 1")), - new Object[] { new Date(firstEvent.getTimeStamp()) }); + new Object[]{ new Date(firstEvent.getTimeStamp()) }); // the last record in the table will be 900th we inserted List all = executeNet(query("select timestamp from %s")).all(); @@ -104,100 +69,47 @@ public class LogMessagesTableTest extends CQLTester assertEquals(loggingEvents.get(900).getTimeStamp(), timestamp.getTime()); } - @Test - public void testMultipleLogsInSameMillisecond() - { - registerVirtualTable(10); - List loggingEvents = getLoggingEvents(10, Instant.now(), 5); - loggingEvents.forEach(table::add); - - // 2 partitions, 5 rows in each - assertEquals(2, numberOfPartitions()); - } - @Test public void testResolvingBufferSize() { LOGS_VIRTUAL_TABLE_MAX_ROWS.setInt(-1); - assertEquals(LogMessagesTable.LOGS_VIRTUAL_TABLE_DEFAULT_ROWS, LogMessagesTable.resolveBufferSize()); + assertEquals(LOGS_VIRTUAL_TABLE_DEFAULT_ROWS, resolveBufferSize()); LOGS_VIRTUAL_TABLE_MAX_ROWS.setInt(0); - assertEquals(LogMessagesTable.LOGS_VIRTUAL_TABLE_DEFAULT_ROWS, LogMessagesTable.resolveBufferSize()); + assertEquals(LOGS_VIRTUAL_TABLE_DEFAULT_ROWS, resolveBufferSize()); LOGS_VIRTUAL_TABLE_MAX_ROWS.setInt(1000001); - assertEquals(LogMessagesTable.LOGS_VIRTUAL_TABLE_DEFAULT_ROWS, LogMessagesTable.resolveBufferSize()); + assertEquals(LOGS_VIRTUAL_TABLE_DEFAULT_ROWS, resolveBufferSize()); LOGS_VIRTUAL_TABLE_MAX_ROWS.setInt(999); - assertEquals(LogMessagesTable.LOGS_VIRTUAL_TABLE_DEFAULT_ROWS, LogMessagesTable.resolveBufferSize()); + assertEquals(999, resolveBufferSize()); LOGS_VIRTUAL_TABLE_MAX_ROWS.setInt(50001); - assertEquals(50001, LogMessagesTable.resolveBufferSize()); + assertEquals(50001, resolveBufferSize()); } - private void registerVirtualTable() + private int resolveBufferSize() { - registerVirtualTable(LogMessagesTable.LOGS_VIRTUAL_TABLE_MIN_ROWS); + return AbstractLoggerVirtualTable.resolveBufferSize(LOGS_VIRTUAL_TABLE_MAX_ROWS.getInt(), + LogMessagesTable.LOGS_VIRTUAL_TABLE_MAX_ROWS, + LOGS_VIRTUAL_TABLE_DEFAULT_ROWS); } - private void registerVirtualTable(int size) + @Override + protected void registerTable(int maxSize) { - table = new LogMessagesTable(keyspace, size); - VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(keyspace, ImmutableList.of(table))); + registerVirtualTable(new LogMessagesTable(keyspace, maxSize)); } - private int numberOfPartitions() + @Override + protected void registerTable() { - DataSet data = table.data(); - - Iterator partitions = data.getPartitions(DataRange.allData(new LocalPartitioner(TimestampType.instance))); - - int numberOfPartitions = 0; - - while (partitions.hasNext()) - { - partitions.next(); - numberOfPartitions += 1; - } - - return numberOfPartitions; + registerTable(1000); } - private String query(String query) + @Override + protected String getMessage(long timestamp) { - return String.format(query, table.toString()); - } - - private List getLoggingEvents(int size) - { - return getLoggingEvents(size, Instant.now(), 1); - } - - private List getLoggingEvents(int size, Instant firstTimestamp, int logsInMillisecond) - { - List logs = new LinkedList<>(); - int partitions = size / logsInMillisecond; - - for (int i = 0; i < partitions; i++) - { - long timestamp = firstTimestamp.toEpochMilli(); - firstTimestamp = firstTimestamp.plusSeconds(1); - - for (int j = 0; j < logsInMillisecond; j++) - logs.add(getLoggingEvent(timestamp)); - } - - return logs; - } - - private LoggingEvent getLoggingEvent(long timestamp) - { - LoggingEvent event = new LoggingEvent(); - event.setLevel(Level.INFO); - event.setMessage("message " + timestamp); - event.setLoggerName("logger " + timestamp); - event.setThreadName(Thread.currentThread().getName()); - event.setTimeStamp(timestamp); - - return event; + return "message " + timestamp; } } diff --git a/test/unit/org/apache/cassandra/db/virtual/SlowQueriesTableTest.java b/test/unit/org/apache/cassandra/db/virtual/SlowQueriesTableTest.java new file mode 100644 index 0000000000..1c261e7cc9 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/virtual/SlowQueriesTableTest.java @@ -0,0 +1,145 @@ +/* + * 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.virtual; + +import java.util.List; +import java.util.Random; + +import org.junit.Test; + +import ch.qos.logback.classic.spi.LoggingEvent; +import org.apache.cassandra.db.monitoring.MonitorableImpl; +import org.apache.cassandra.db.monitoring.MonitoringTask; +import org.apache.cassandra.db.monitoring.MonitoringTask.Operation; +import org.apache.cassandra.utils.Generators; +import org.quicktheories.impl.JavaRandom; + +import static org.apache.cassandra.config.CassandraRelevantProperties.LOGS_SLOW_QUERIES_VIRTUAL_TABLE_MAX_ROWS; +import static org.apache.cassandra.db.virtual.SlowQueriesTable.LOGS_VIRTUAL_TABLE_DEFAULT_ROWS; +import static org.apache.cassandra.db.virtual.SlowQueriesTable.LOGS_VIRTUAL_TABLE_MAX_ROWS; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public class SlowQueriesTableTest extends AbstractLoggerVirtualTableTest +{ + private final Random random = new Random(); + private final JavaRandom javaRandom = new JavaRandom(random); + + @Override + protected void registerTable(int maxSize) + { + registerVirtualTable(new SlowQueriesTable(keyspace, maxSize)); + } + + @Override + protected void registerTable() + { + registerTable(1000); + } + + @Test + public void testLimitedCapacity() + { + registerTable(100); + + int numberOfRows = 1000; + List loggingEvents = getLoggingEvents(numberOfRows); + assertEquals(1000, loggingEvents.size()); + loggingEvents.forEach(table::add); + + // even we inserted 1000 rows, only 100 are present as its capacity is bounded + assertEquals(100, executeNet(query("select * from %s")).all().size()); + } + + @Test + public void testDelete() + { + registerTable(); + + int numberOfRows = 100; + List loggingEvents = getLoggingEvents(numberOfRows); + loggingEvents.forEach(table::add); + + Operation operation = table.buffer.get(0); + + assertEquals(100, executeNet(query("select * from %s")).all().size()); + execute(query("delete from %s where keyspace_name = '" + operation.keyspace() + '\'')); + assertTrue(executeNet(query("select * from %s")).all().size() < 100); + } + + @Test + public void testResolvingBufferSize() + { + LOGS_SLOW_QUERIES_VIRTUAL_TABLE_MAX_ROWS.setInt(-1); + assertEquals(LOGS_VIRTUAL_TABLE_DEFAULT_ROWS, resolveBufferSize()); + + LOGS_SLOW_QUERIES_VIRTUAL_TABLE_MAX_ROWS.setInt(0); + assertEquals(LOGS_VIRTUAL_TABLE_DEFAULT_ROWS, resolveBufferSize()); + + LOGS_SLOW_QUERIES_VIRTUAL_TABLE_MAX_ROWS.setInt(1000001); + assertEquals(LOGS_VIRTUAL_TABLE_DEFAULT_ROWS, resolveBufferSize()); + + LOGS_SLOW_QUERIES_VIRTUAL_TABLE_MAX_ROWS.setInt(999); + assertEquals(999, resolveBufferSize()); + + LOGS_SLOW_QUERIES_VIRTUAL_TABLE_MAX_ROWS.setInt(50001); + assertEquals(50001, resolveBufferSize()); + } + + private int resolveBufferSize() + { + return AbstractLoggerVirtualTable.resolveBufferSize(LOGS_SLOW_QUERIES_VIRTUAL_TABLE_MAX_ROWS.getInt(), + LOGS_VIRTUAL_TABLE_MAX_ROWS, + LOGS_VIRTUAL_TABLE_DEFAULT_ROWS); + } + + + @Override + protected String getMessage(long timestamp) + { + MonitoringTask.SlowOperation slowOperation = new MonitoringTask.SlowOperation(new MonitorableImpl() + { + @Override + public String name() + { + return Generators.SYMBOL_GEN.generate(javaRandom); + } + + @Override + public String monitoredOnKeyspace() + { + return Generators.SYMBOL_GEN.generate(javaRandom); + } + + @Override + public String monitoredOnTable() + { + return Generators.SYMBOL_GEN.generate(javaRandom); + } + + @Override + public boolean isCrossNode() + { + return random.nextBoolean(); + } + }, timestamp); + + return Operation.serialize(List.of(slowOperation)); + } +}