CASSANDRA-13001 Implement appender of slow queries to system_views.slow_queries table

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/auth/MutualTlsAuthenticator.java

@@ -44,7 +44,7 @@
 import org.apache.cassandra.utils.NoSpamLogger;
 
 import static org.apache.cassandra.auth.IAuthenticator.AuthenticationMode.MTLS;
-import static org.apache.cassandra.config.EncryptionOptions.ClientAuth.REQUIRED;
+import static org.apache.cassandra.config.EncryptionOptions.ClientEncryptionOptions.ClientAuth.REQUIRED;
 
 /**
  * Performs mTLS authentication for client connections by extracting identities from client certificate

src/java/org/apache/cassandra/auth/MutualTlsInternodeAuthenticator.java

@@ -49,7 +49,7 @@
 import org.apache.cassandra.metrics.MutualTlsMetrics;
 import org.apache.cassandra.utils.NoSpamLogger;
 
-import static org.apache.cassandra.config.EncryptionOptions.ClientAuth.REQUIRED;
+import static org.apache.cassandra.config.EncryptionOptions.ClientEncryptionOptions.ClientAuth.REQUIRED;
 
 /**
  * Performs mTLS authentication for internode connections by extracting identities from the certificates of incoming

src/java/org/apache/cassandra/auth/MutualTlsWithPasswordFallbackAuthenticator.java

@@ -81,7 +81,7 @@ public SaslNegotiator newSaslNegotiator(InetAddress clientAddress, Certificate[]
     public void validateConfiguration() throws ConfigurationException
     {
         Config config = DatabaseDescriptor.getRawConfig();
-        if (config.client_encryption_options.getClientAuth() == EncryptionOptions.ClientAuth.NOT_REQUIRED)
+        if (config.client_encryption_options.getClientAuth() == EncryptionOptions.ClientEncryptionOptions.ClientAuth.NOT_REQUIRED)
         {
             String msg = "MutualTlsWithPasswordFallbackAuthenticator requires client_encryption_options.require_client_auth to be optional/true";
             throw new ConfigurationException(msg);

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

@@ -27,6 +27,7 @@
 
 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/config/Config.java

@@ -454,7 +454,7 @@ public static class SSTableConfig
     public String failure_detector = "FailureDetector";
 
     public EncryptionOptions.ServerEncryptionOptions server_encryption_options = new EncryptionOptions.ServerEncryptionOptions();
-    public EncryptionOptions client_encryption_options = new EncryptionOptions();
+    public EncryptionOptions.ClientEncryptionOptions client_encryption_options = new EncryptionOptions.ClientEncryptionOptions();
 
     public JMXServerOptions jmx_server_options;
 

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

@@ -156,7 +156,7 @@
 import static org.apache.cassandra.config.DataRateSpec.DataRateUnit.BYTES_PER_SECOND;
 import static org.apache.cassandra.config.DataRateSpec.DataRateUnit.MEBIBYTES_PER_SECOND;
 import static org.apache.cassandra.config.DataStorageSpec.DataStorageUnit.MEBIBYTES;
-import static org.apache.cassandra.config.EncryptionOptions.ClientAuth.REQUIRED;
+import static org.apache.cassandra.config.EncryptionOptions.ClientEncryptionOptions.ClientAuth.REQUIRED;
 import static org.apache.cassandra.db.ConsistencyLevel.ALL;
 import static org.apache.cassandra.db.ConsistencyLevel.EACH_QUORUM;
 import static org.apache.cassandra.db.ConsistencyLevel.LOCAL_QUORUM;
@@ -1088,10 +1088,10 @@ else if (conf.commitlog_segment_size.toKibibytes() < 2 * conf.max_mutation_size.
         }
         else if (JMXServerOptions.isEnabledBySystemProperties())
         {
-            throw new ConfigurationException("Configure either jmx_server_options in cassandra.yaml and comment out " +
-                                             "configure_jmx function call in cassandra-env.sh or keep cassandra-env.sh " +
-                                             "to call configure_jmx function but you have to keep jmx_server_options " +
-                                             "in cassandra.yaml commented out.");
+                throw new ConfigurationException("Configure either jmx_server_options in cassandra.yaml and comment out " +
+                                                 "configure_jmx function call in cassandra-env.sh or keep cassandra-env.sh " +
+                                                 "to call configure_jmx function but you have to keep jmx_server_options " +
+                                                 "in cassandra.yaml commented out.");
         }
 
         conf.jmx_server_options.jmx_encryption_options.applyConfig();
@@ -3973,7 +3973,7 @@ public static void setInternodeMessagingEncyptionOptions(EncryptionOptions.Serve
         conf.server_encryption_options = encryptionOptions;
     }
 
-    public static EncryptionOptions getNativeProtocolEncryptionOptions()
+    public static EncryptionOptions.ClientEncryptionOptions getNativeProtocolEncryptionOptions()
     {
         return conf.client_encryption_options;
     }
@@ -3984,7 +3984,7 @@ public static JMXServerOptions getJmxServerOptions()
     }
 
     @VisibleForTesting
-    public static void updateNativeProtocolEncryptionOptions(Function<EncryptionOptions, EncryptionOptions> update)
+    public static void updateNativeProtocolEncryptionOptions(Function<EncryptionOptions.ClientEncryptionOptions, EncryptionOptions.ClientEncryptionOptions> update)
     {
         conf.client_encryption_options = update.apply(conf.client_encryption_options);
     }