Implement appender of slow queries to system_views.slow_queries table

patch by Stefan Miklosovic; reviewed by Dmitry Konstantinov, Bernardo Botella for CASSANDRA-13001

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)

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
 

conf/logback.xml

@@ -119,6 +119,18 @@ appender reference in the root level section below.
     </filter>
   </appender> -->
 
+  <!-- Uncomment below configuration and corresponding appender-ref in slow_queries logger to activate
+       logging into system_views.slow_queries virtual table. -->
+  <!-- <appender name="SLOW_QUERIES_APPENDER" class="org.apache.cassandra.utils.logging.SlowQueriesAppender"/> -->
+
+  <!-- Log slow queries to system_views.slow_queries virtual table -->
+  <logger name="slow_queries" additivity="false" level="DEBUG">
+    <appender-ref ref="DEBUGLOG"/>
+    <!-- uncomment this appender reference together with appender definition above
+         to start to put slow queries into system_views.slow_queries virtual table -->
+    <!-- <appender-ref ref="SLOW_QUERIES_APPENDER"/> -->
+  </logger>
+
   <root level="INFO">
     <appender-ref ref="SYSTEMLOG" />
     <appender-ref ref="STDOUT" />

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.
 <pattern>%-5level [%thread] %date\{ISO8601} %F:%L - %msg%n</pattern>
 </encoder>
 
-=== 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]
+----
+  <!-- Uncomment below configuration and corresponding appender-ref in slow_queries logger to activate
+       logging into system_views.slow_queries virtual table. -->
+  <!-- <appender name="SLOW_QUERIES_APPENDER" class="org.apache.cassandra.utils.logging.SlowQueriesAppender"/> -->
+
+  <!-- Log slow queries to system_views.slow_queries virtual table -->
+  <logger name="slow_queries" additivity="false" level="DEBUG">
+    <appender-ref ref="DEBUGLOG"/>
+    <!-- uncomment this appender reference together with appender definition above
+         to start to put slow queries into system_views.slow_queries virtual table
+    <appender-ref ref="SLOW_QUERIES_APPENDER"/>
+    -->
+  </logger>
+----
+
+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]

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)),
     /**

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;
+    }
 }

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;
+        }
+    }
 }

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<Operation> 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;
+            minTimeNanos = totalTimeNanos;
-            maxTime = 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++;
+            if (operation != null)
-            totalTimeNanos += operation.totalTimeNanos;
+            {
-            maxTime = Math.max(maxTime, operation.maxTime);
+                String monitored = operation.monitoredOnKeyspace();
-            minTime = Math.min(minTime, operation.minTime);
+                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<Operation> operations)
+        {
+            return JsonUtils.writeAsJsonString(operations);
+        }
+
+        private static final TypeReference<List<Operation>> TYPE_REFERENCE = new TypeReference<>() {};
+
+        public static List<Operation> 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(minTimeNanos),
-                                     NANOSECONDS.toMillis(maxTime),
+                                     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(minTimeNanos),
-                                     NANOSECONDS.toMillis(maxTime),
+                                     NANOSECONDS.toMillis(maxTimeNanos),
                                      NANOSECONDS.toMillis(operation.slowTimeoutNanos()),
                                      operation.isCrossNode() ? "msec/cross-node" : "msec");
         }

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 <U> type parameter saying what object is stored in internal bounded list for query purposes
+ */
+public abstract class AbstractLoggerVirtualTable<U> extends AbstractMutableVirtualTable
+{
+    private static final Logger logger = LoggerFactory.getLogger(AbstractLoggerVirtualTable.class);
+
+    // please be sure operations on this structure are thread-safe
+    protected final List<U> 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<U> 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<U> getMessages(LoggingEvent event);
+
+    @Override
+    public void truncate()
+    {
+        synchronized (buffer)
+        {
+            buffer.clear();
+        }
+    }
+
+    @Override
+    public boolean allowFilteringImplicitly()
+    {
+        return false;
+    }
+
+    private static final class BoundedLinkedList<T> extends LinkedList<T>
+    {
+        private final int maxSize;
+
+        public static <T> List<T> 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;
+        }
+    }
+}

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 <a href="https://issues.apache.org/jira/browse/CASSANDRA-18238">CASSANDRA-18238</a>
  * @see org.apache.cassandra.utils.logging.VirtualTableAppender
  */
-public final class LogMessagesTable extends AbstractMutableVirtualTable
+public final class LogMessagesTable extends AbstractLoggerVirtualTable<LoggingEvent>
 {
-    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<LogMessage> 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);
+    @Override
-        buffer = BoundedLinkedList.create(size);
+    public List<LoggingEvent> getMessages(LoggingEvent event)
+    {
+        return List.of(event);
     }
 
     @Override
@@ -103,12 +100,12 @@ public final class LogMessagesTable extends AbstractMutableVirtualTable
 
             int index = 0;
 
-            Iterator<LogMessage> iterator = buffer.listIterator();
+            Iterator<LoggingEvent> 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)
+                result.row(new Date(milliSecondsOfCurrentLog), index)
-                      .column(LOGGER_COLUMN_NAME, log.logger)
+                      .column(LOGGER_COLUMN_NAME, log.getLoggerName())
-                      .column(LEVEL_COLUMN_NAME, log.level)
+                      .column(LEVEL_COLUMN_NAME, log.getLevel().toString())
-                      .column(MESSAGE_COLUMN_NAME, log.message);
+                      .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<T> extends LinkedList<T>
-    {
-        private final int maxSize;
-
-        public static <T> List<T> 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;
-        }
-    }
 }

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<Operation>
+{
+    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<Operation> 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<Operation> getMessages(LoggingEvent event)
+    {
+        try
+        {
+            List<Operation> 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;
+    }
+}

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))

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
+        // Flush log messages to system_views.system_logs virtual table as there were messages already logged
-        // before that virtual table was instantiated
+        // 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()

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<LoggingEvent>
+{
+    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<LoggingEvent> messageBuffer = new LinkedList<>();
+
+    protected <T> T getVirtualTable(Class<T> vtableClass, String tableName)
+    {
+        VirtualKeyspace keyspace = VirtualKeyspaceRegistry.instance.getKeyspaceNullable(VIRTUAL_VIEWS);
+
+        if (keyspace == null)
+            return null;
+
+        Optional<VirtualTable> 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<? extends AbstractLoggerVirtualTable<?>> 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);
+        }
+    }
+}

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<Appender<?>> getAppender(Class<?> appenderClass, String name)
+    public Optional<Logger> 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 <T extends Appender<?>> Optional<T> getAppender(Class<T> appenderClass, String appenderName)
     {
         LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
         for (Logger logBackLogger : lc.getLoggerList())
@@ -146,15 +160,15 @@ public class LogbackLoggingSupport implements LoggingSupport
             for (Iterator<Appender<ILoggingEvent>> iterator = logBackLogger.iteratorForAppenders(); iterator.hasNext();)
             {
                 Appender<ILoggingEvent> appender = iterator.next();
-                if (appender.getClass() == appenderClass && appender.getName().equals(name))
+                if (appender.getClass() == appenderClass && appender.getName().equals(appenderName))
-                    return Optional.of(appender);
+                    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<Appender<ILoggingEvent>> 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)

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<String, String> getLoggingLevels();
 
-    default Optional<Appender<?>> getAppender(Class<?> appenderClass, String appenderName)
+    default <T extends Appender<?>> Optional<T> getAppender(Class<T> appenderClass, String appenderName)
+    {
+        return Optional.empty();
+    }
+
+    default Optional<Logger> getLogger(String loggerName)
     {
         return Optional.empty();
     }

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);
+    }
+}

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<LoggingEvent>
+public final class VirtualTableAppender extends AbstractVirtualTableAppender
 {
     public static final String APPENDER_NAME = "CQLLOG";
 
     private static final Set<String> forbiddenLoggers = ImmutableSet.of(FileAuditLogger.class.getName());
 
-    private LogMessagesTable logs;
+    private AbstractLoggerVirtualTable<?> logs;
 
-    // for holding messages until virtual registry contains logs virtual table
+    public VirtualTableAppender()
-    // as it takes some time during startup of a node to initialise virtual tables but messages are
+    {
-    // logged already
+        super(LogMessagesTable.LOGS_VIRTUAL_TABLE_DEFAULT_ROWS);
-    private final List<LoggingEvent> messageBuffer = new LinkedList<>();
+    }
 
     @Override
     protected void append(LoggingEvent eventObject)
     {
         if (!forbiddenLoggers.contains(eventObject.getLoggerName()))
-        {
+            logs = appendToVirtualTable(logs, eventObject, LogMessagesTable.TABLE_NAME);
-            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<VirtualTable> 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);
     }
 }

test/conf/logback-dtest_with_slow_query_appender.xml

@@ -0,0 +1,63 @@
+<!--
+  ~ 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.
+  -->
+
+<configuration debug="false" scan="true" scanPeriod="60 seconds">
+    <define name="cluster_id" class="org.apache.cassandra.distributed.impl.ClusterIDDefiner" />
+    <define name="instance_id" class="org.apache.cassandra.distributed.impl.InstanceIDDefiner" />
+
+    <!-- Shutdown hook ensures that async appender flushes -->
+    <shutdownHook class="ch.qos.logback.core.hook.DelayingShutdownHook"/>
+
+    <appender name="INSTANCEFILE" class="ch.qos.logback.core.FileAppender">
+        <file>./build/test/logs/${cassandra.testtag}/${suitename}/${cluster_id}/${instance_id}/system.log</file>
+        <encoder>
+            <pattern>%-5level [%thread] ${instance_id} %date{ISO8601} %msg%n</pattern>
+        </encoder>
+        <immediateFlush>true</immediateFlush>
+    </appender>
+
+    <appender name="INSTANCESTDERR" target="System.err" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%-5level %date{HH:mm:ss,SSS} %msg%n</pattern>
+        </encoder>
+        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
+            <level>WARN</level>
+        </filter>
+    </appender>
+
+    <appender name="INSTANCESTDOUT" target="System.out" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%-5level [%thread] ${instance_id} %date{ISO8601} %F:%L - %msg%n</pattern>
+        </encoder>
+        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
+            <level>DEBUG</level>
+        </filter>
+    </appender>
+
+    <appender name="SLOW_QUERIES_APPENDER" class="org.apache.cassandra.utils.logging.SlowQueriesAppender"/>
+
+    <logger name="slow_operations" additivity="false" level="DEBUG">
+        <appender-ref ref="SLOW_QUERIES_APPENDER"/>
+    </logger>
+
+    <root level="DEBUG">
+        <appender-ref ref="INSTANCEFILE" /> <!-- use blocking to avoid race conditions with appending and searching -->
+        <appender-ref ref="INSTANCESTDERR" />
+        <appender-ref ref="INSTANCESTDOUT" />
+    </root>
+</configuration>

test/conf/logback-dtest_with_slow_query_appender_invalid.xml

@@ -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.
+  -->
+
+<configuration debug="false" scan="true" scanPeriod="60 seconds">
+    <define name="cluster_id" class="org.apache.cassandra.distributed.impl.ClusterIDDefiner" />
+    <define name="instance_id" class="org.apache.cassandra.distributed.impl.InstanceIDDefiner" />
+
+    <!-- Shutdown hook ensures that async appender flushes -->
+    <shutdownHook class="ch.qos.logback.core.hook.DelayingShutdownHook"/>
+
+    <appender name="INSTANCEFILE" class="ch.qos.logback.core.FileAppender">
+        <file>./build/test/logs/${cassandra.testtag}/${suitename}/${cluster_id}/${instance_id}/system.log</file>
+        <encoder>
+            <pattern>%-5level [%thread] ${instance_id} %date{ISO8601} %msg%n</pattern>
+        </encoder>
+        <immediateFlush>true</immediateFlush>
+    </appender>
+
+    <appender name="INSTANCESTDERR" target="System.err" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%-5level %date{HH:mm:ss,SSS} %msg%n</pattern>
+        </encoder>
+        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
+            <level>WARN</level>
+        </filter>
+    </appender>
+
+    <appender name="INSTANCESTDOUT" target="System.out" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%-5level [%thread] ${instance_id} %date{ISO8601} %F:%L - %msg%n</pattern>
+        </encoder>
+        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
+            <level>DEBUG</level>
+        </filter>
+    </appender>
+
+    <appender name="SLOW_QUERIES_APPENDER" class="org.apache.cassandra.utils.logging.SlowQueriesAppender"/>
+    <appender name="SLOW_QUERIES_APPENDER_2" class="org.apache.cassandra.utils.logging.SlowQueriesAppender"/>
+
+    <appender name="CQLLOG" class="org.apache.cassandra.utils.logging.VirtualTableAppender">
+        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
+            <level>INFO</level>
+        </filter>
+    </appender>
+
+    <!-- multiple appenders are not allowed -->
+    <logger name="slow_operations" additivity="false" level="DEBUG">
+        <appender-ref ref="SLOW_QUERIES_APPENDER"/>
+        <appender-ref ref="SLOW_QUERIES_APPENDER_2"/>
+    </logger>
+
+    <root level="DEBUG">
+        <appender-ref ref="INSTANCEFILE" /> <!-- use blocking to avoid race conditions with appending and searching -->
+        <appender-ref ref="INSTANCESTDERR" />
+        <appender-ref ref="INSTANCESTDOUT" />
+        <appender-ref ref="CQLLOG" />
+    </root>
+</configuration>

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();
+}

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);
+    }
+}

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<MonitoringTask.Operation> deserialize = MonitoringTask.Operation.deserialize(serialize);
+    }
+}

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<TestingLogMessage> rows = getRows(cluster);
+            List<LogMessage> 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<TestingLogMessage> getRows(Cluster cluster)
+    private List<LogMessage> getRows(Cluster cluster)
     {
         SimpleQueryResult simpleQueryResult = cluster.coordinator(1).executeWithResult(query("select * from %s"), ONE);
-        List<TestingLogMessage> rows = new ArrayList<>();
+        List<LogMessage> 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()
+        public LogMessage(long timestamp, String logger, String level, String message, int order)
-    {
-        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)
         {
-            super(timestamp, logger, level, message);
+            this.timestamp = timestamp;
+            this.logger = logger;
+            this.level = level;
+            this.message = message;
             this.order = order;
         }
     }

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<U> extends CQLTester
+{
+    protected final String keyspace = createKeyspaceName();
+
+    protected AbstractLoggerVirtualTable<U> table;
+
+    @Test
+    public void testTruncate()
+    {
+        registerTable();
+
+        int numberOfRows = 100;
+        List<LoggingEvent> 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<LoggingEvent> 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<LoggingEvent> 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<Row> 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<U> 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<LoggingEvent> getLoggingEvents(int size)
+    {
+        return getLoggingEvents(size, Instant.now(), 1);
+    }
+
+    protected List<LoggingEvent> getLoggingEvents(int size, Instant firstTimestamp, int logsInMillisecond)
+    {
+        List<LoggingEvent> 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<AbstractVirtualTable.Partition> 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);
+}

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<LoggingEvent>
 {
-    private String keyspace = createKeyspaceName();
-    private LogMessagesTable table;
-
     @Test
-    public void testTruncate() throws Throwable
+    public void testMultipleLogsInSameMillisecond()
     {
-        registerVirtualTable();
+        registerTable();
-
+        List<LoggingEvent> loggingEvents = getLoggingEvents(10, Instant.now(), 5);
-        int numberOfRows = 100;
-        List<LoggingEvent> loggingEvents = getLoggingEvents(numberOfRows);
         loggingEvents.forEach(table::add);
 
-        execute(query("truncate %s"));
+        // 2 partitions, 5 rows in each
-
+        assertEquals(2, numberOfPartitions());
-        assertTrue(executeNet(query("select timestamp from %s")).all().isEmpty());
     }
 
     @Test
-    public void empty() throws Throwable
+    public void testLimitedCapacity()
     {
-        registerVirtualTable();
+        registerTable(100);
-        assertEmpty(execute(query("select * from %s")));
-    }
-
-    @Test
-    public void testInsert()
-    {
-        registerVirtualTable();
-
-        int numberOfRows = 1000;
-        List<LoggingEvent> loggingEvents = getLoggingEvents(numberOfRows);
-        loggingEvents.forEach(table::add);
-
-        assertEquals(numberOfRows, numberOfPartitions());
-    }
-
-    @Test
-    public void testLimitedCapacity() throws Throwable
-    {
-        registerVirtualTable(100);
 
         int numberOfRows = 1000;
         List<LoggingEvent> 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<Row> 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<LoggingEvent> 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);
+        registerVirtualTable(new LogMessagesTable(keyspace, maxSize));
-        VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(keyspace, ImmutableList.of(table)));
     }
 
-    private int numberOfPartitions()
+    @Override
+    protected void registerTable()
     {
-        DataSet data = table.data();
+        registerTable(1000);
-
-        Iterator<Partition> partitions = data.getPartitions(DataRange.allData(new LocalPartitioner(TimestampType.instance)));
-
-        int numberOfPartitions = 0;
-
-        while (partitions.hasNext())
-        {
-            partitions.next();
-            numberOfPartitions += 1;
-        }
-
-        return numberOfPartitions;
     }
 
-    private String query(String query)
+    @Override
+    protected String getMessage(long timestamp)
     {
-        return String.format(query, table.toString());
+        return "message " + timestamp;
-    }
-
-    private List<LoggingEvent> getLoggingEvents(int size)
-    {
-        return getLoggingEvents(size, Instant.now(), 1);
-    }
-
-    private List<LoggingEvent> getLoggingEvents(int size, Instant firstTimestamp, int logsInMillisecond)
-    {
-        List<LoggingEvent> 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;
     }
 }

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<Operation>
+{
+    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<LoggingEvent> 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<LoggingEvent> 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));
+    }
+}