review

Author: smiklosovic

Diff for: conf/logback.xml

@@ -124,9 +124,13 @@ appender reference in the root level section below.
   <!-- <appender name="SLOW_QUERIES_TO_VTABLE" 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">
+  <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_TO_VTABLE"/>
-  </logger> -->
+    -->
+  </logger>
 
   <root level="INFO">
     <appender-ref ref="SYSTEMLOG" />

Diff for: src/java/org/apache/cassandra/db/monitoring/MonitoringTask.java

@@ -192,14 +192,15 @@ boolean logSlowOperations(long approxCurrentTimeNanos)
         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 String's to debug method (or objects, on which toString()
+                    // 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.
@@ -370,10 +371,10 @@ public abstract static class Operation
         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 */
@@ -385,10 +386,10 @@ public abstract static class Operation
          * which does not follow wall clock and is useless for
          * reporting purposes e.g. in virtual tables
          */
-        private final long timestamp;
+        private final long timestampMs;
 
         // optional keyspace and table this operation acts on
-        // used upon deserialisation
+        // used upon deserialization
         private String keyspace;
         private String table;
         private boolean crossNode;
@@ -398,17 +399,17 @@ public abstract static class Operation
             this.operation = operation;
             numTimesReported = 1;
             totalTimeNanos = failedAtNanos - operation.creationTimeNanos();
-            minTime = totalTimeNanos;
-            maxTime = totalTimeNanos;
-            timestamp = Clock.Global.currentTimeMillis();
+            minTimeNanos = totalTimeNanos;
+            maxTimeNanos = totalTimeNanos;
+            timestampMs = Clock.Global.currentTimeMillis() - (Clock.Global.nanoTime() - operation.creationTimeNanos()) / 1_000_000;
         }
 
         void add(Operation operation)
         {
             numTimesReported++;
             totalTimeNanos += operation.totalTimeNanos;
-            maxTime = Math.max(maxTime, operation.maxTime);
-            minTime = Math.min(minTime, operation.minTime);
+            maxTimeNanos = Math.max(maxTimeNanos, operation.maxTimeNanos);
+            minTimeNanos = Math.min(minTimeNanos, operation.minTimeNanos);
         }
 
         public abstract String getLogMessage();
@@ -467,15 +468,15 @@ public long totalTimeNanos()
         }
 
         @JsonProperty
-        public long maxTime()
+        public long maxTimeNanos()
         {
-            return maxTime;
+            return maxTimeNanos;
         }
 
         @JsonProperty
-        public long minTime()
+        public long minTimeNanos()
         {
-            return minTime;
+            return minTimeNanos;
         }
 
         @JsonIgnore
@@ -485,19 +486,21 @@ public long averageTime()
         }
 
         @JsonProperty
-        public long timestamp()
+        public long timestampMs()
         {
-            return timestamp;
+            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, new TypeReference<>() {});
+            return JsonUtils.JSON_OBJECT_MAPPER.readValue(message, TYPE_REFERENCE);
         }
     }
 
@@ -524,8 +527,8 @@ public String getLogMessage()
                                      name(),
                                      numTimesReported,
                                      NANOSECONDS.toMillis(totalTimeNanos / numTimesReported),
-                                     NANOSECONDS.toMillis(minTime),
-                                     NANOSECONDS.toMillis(maxTime),
+                                     NANOSECONDS.toMillis(minTimeNanos),
+                                     NANOSECONDS.toMillis(maxTimeNanos),
                                      NANOSECONDS.toMillis(operation.timeoutNanos()),
                                      operation.isCrossNode() ? "msec/cross-node" : "msec");
         }
@@ -537,7 +540,7 @@ public String getLogMessage()
     @VisibleForTesting
     public final static class SlowOperation extends Operation
     {
-        // purely for deserialisation purposes
+        // purely for deserialization purposes
         public SlowOperation()
         {
             this(Monitorable.NO_OP, 0);
@@ -562,8 +565,8 @@ public String getLogMessage()
                                      name(),
                                      numTimesReported,
                                      NANOSECONDS.toMillis(totalTimeNanos/ numTimesReported),
-                                     NANOSECONDS.toMillis(minTime),
-                                     NANOSECONDS.toMillis(maxTime),
+                                     NANOSECONDS.toMillis(minTimeNanos),
+                                     NANOSECONDS.toMillis(maxTimeNanos),
                                      NANOSECONDS.toMillis(operation.slowTimeoutNanos()),
                                      operation.isCrossNode() ? "msec/cross-node" : "msec");
         }

Diff for: src/java/org/apache/cassandra/db/virtual/AbstractLoggerVirtualTable.java

@@ -38,12 +38,13 @@ public abstract class AbstractLoggerVirtualTable<U> extends AbstractMutableVirtu
 {
     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 min, int max, int defaultSize)
+    protected static int resolveBufferSize(int wantedSize, int max, int defaultSize)
     {
-        return (wantedSize < min || wantedSize > max) ? defaultSize : wantedSize;
+        return (wantedSize < 0 || wantedSize > max) ? defaultSize : wantedSize;
     }
 
     protected AbstractLoggerVirtualTable(TableMetadata metadata, int maxSize)
@@ -70,7 +71,10 @@ public void add(LoggingEvent event)
     @Override
     public void truncate()
     {
-        buffer.clear();
+        synchronized (buffer)
+        {
+            buffer.clear();
+        }
     }
 
     @Override
@@ -94,7 +98,7 @@ private BoundedLinkedList(int maxSize)
         }
 
         @Override
-        public boolean add(T t)
+        public synchronized boolean add(T t)
         {
             if (size() == maxSize)
                 removeLast();

Diff for: src/java/org/apache/cassandra/db/virtual/LogMessagesTable.java

@@ -48,7 +48,6 @@
  */
 public final class LogMessagesTable extends AbstractLoggerVirtualTable<LoggingEvent>
 {
-    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;
 
@@ -64,7 +63,6 @@ public final class LogMessagesTable extends AbstractLoggerVirtualTable<LoggingEv
     LogMessagesTable(String keyspace)
     {
         this(keyspace, resolveBufferSize(CassandraRelevantProperties.LOGS_VIRTUAL_TABLE_MAX_ROWS.getInt(),
-                                         LOGS_VIRTUAL_TABLE_MIN_ROWS,
                                          LOGS_VIRTUAL_TABLE_MAX_ROWS,
                                          LOGS_VIRTUAL_TABLE_DEFAULT_ROWS));
     }

Diff for: src/java/org/apache/cassandra/db/virtual/SlowQueriesTable.java

@@ -45,7 +45,6 @@ public class SlowQueriesTable extends AbstractLoggerVirtualTable<Operation>
 {
     private static final Logger logger = LoggerFactory.getLogger(SlowQueriesTable.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;
 
@@ -65,7 +64,6 @@ public class SlowQueriesTable extends AbstractLoggerVirtualTable<Operation>
     SlowQueriesTable(String keyspace)
     {
         this(keyspace, resolveBufferSize(CassandraRelevantProperties.LOGS_SLOW_QUERIES_VIRTUAL_TABLE_MAX_ROWS.getInt(),
-                                         LOGS_VIRTUAL_TABLE_MIN_ROWS,
                                          LOGS_VIRTUAL_TABLE_MAX_ROWS,
                                          LOGS_VIRTUAL_TABLE_DEFAULT_ROWS));
     }
@@ -121,9 +119,9 @@ public DataSet data()
             {
                 Operation operation = iterator.next();
 
-                result.row(operation.keyspace(), operation.table(), new Date(operation.timestamp()), operation.name())
-                      .column(MINIMUM_TIME_COLUMN_NAME, NANOSECONDS.toMillis(operation.minTime()))
-                      .column(MAXIMUM_TIME_COLUMN_NAME, NANOSECONDS.toMillis(operation.maxTime()))
+                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());
@@ -144,6 +142,9 @@ public List<Operation> getMessages(LoggingEvent event)
                 String keyspace = operation.keyspace();
                 String table = operation.table();
 
+                // 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 (keyspace == null || table == null)
                     continue;
 

Diff for: src/java/org/apache/cassandra/utils/logging/AbstractVirtualTableAppender.java

@@ -108,8 +108,11 @@ protected AbstractLoggerVirtualTable<?> appendToVirtualTable(AbstractLoggerVirtu
     @Override
     public void stop()
     {
-        messageBuffer.clear();
-        super.stop();
+        synchronized (messageBuffer)
+        {
+            messageBuffer.clear();
+            super.stop();
+        }
     }
 
     /**
@@ -119,17 +122,23 @@ public void stop()
      */
     public void flushBuffer(Class<? extends AbstractLoggerVirtualTable<?>> vtableClass, String tableName)
     {
-        Optional.ofNullable(getVirtualTable(vtableClass, tableName)).ifPresent(vtable -> {
-            messageBuffer.forEach(vtable::add);
-            messageBuffer.clear();
-        });
+        synchronized (messageBuffer)
+        {
+            Optional.ofNullable(getVirtualTable(vtableClass, tableName)).ifPresent(vtable -> {
+                messageBuffer.forEach(vtable::add);
+                messageBuffer.clear();
+            });
+        }
     }
 
     protected 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() < defaultRows)
-            messageBuffer.add(eventObject);
+        synchronized (messageBuffer)
+        {
+            // we restrict how many logging events we can put into buffer,
+            // so we are not growing without any bound when things go south
+            if (messageBuffer.size() < defaultRows)
+                messageBuffer.add(eventObject);
+        }
     }
 }

Diff for: src/java/org/apache/cassandra/utils/logging/SlowQueriesAppender.java

@@ -36,6 +36,10 @@ public SlowQueriesAppender()
     @Override
     protected void append(LoggingEvent eventObject)
     {
+        // slowQueries will be null as long as virtual tables
+        // are not registered, and we already try to put queries there.
+        // As soon as vtable is registered (as part of node's startup / initialisation),
+        // slow queries will never be null again
         slowQueries = appendToVirtualTable(slowQueries, eventObject, SlowQueriesTable.TABLE_NAME);
     }
 }

Diff for: test/unit/org/apache/cassandra/db/virtual/LogMessagesTableTest.java

@@ -77,13 +77,13 @@ public void testResolvingBufferSize()
         assertEquals(LogMessagesTable.LOGS_VIRTUAL_TABLE_DEFAULT_ROWS, resolveBufferSize());
 
         LOGS_VIRTUAL_TABLE_MAX_ROWS.setInt(0);
-        assertEquals(LogMessagesTable.LOGS_VIRTUAL_TABLE_DEFAULT_ROWS, resolveBufferSize());
+        assertEquals(0, resolveBufferSize());
 
         LOGS_VIRTUAL_TABLE_MAX_ROWS.setInt(1000001);
         assertEquals(LogMessagesTable.LOGS_VIRTUAL_TABLE_DEFAULT_ROWS, resolveBufferSize());
 
         LOGS_VIRTUAL_TABLE_MAX_ROWS.setInt(999);
-        assertEquals(LogMessagesTable.LOGS_VIRTUAL_TABLE_DEFAULT_ROWS, resolveBufferSize());
+        assertEquals(999, resolveBufferSize());
 
         LOGS_VIRTUAL_TABLE_MAX_ROWS.setInt(50001);
         assertEquals(50001, resolveBufferSize());
@@ -92,7 +92,6 @@ public void testResolvingBufferSize()
     private int resolveBufferSize()
     {
         return AbstractLoggerVirtualTable.resolveBufferSize(CassandraRelevantProperties.LOGS_VIRTUAL_TABLE_MAX_ROWS.getInt(),
-                                                            LogMessagesTable.LOGS_VIRTUAL_TABLE_MIN_ROWS,
                                                             LogMessagesTable.LOGS_VIRTUAL_TABLE_MAX_ROWS,
                                                             LogMessagesTable.LOGS_VIRTUAL_TABLE_DEFAULT_ROWS);
     }
@@ -106,7 +105,7 @@ protected void registerTable(int maxSize)
     @Override
     protected void registerTable()
     {
-        registerTable(LogMessagesTable.LOGS_VIRTUAL_TABLE_MIN_ROWS);
+        registerTable(1000);
     }
 
     @Override

Diff for: test/unit/org/apache/cassandra/db/virtual/SlowQueriesTableTest.java

@@ -48,7 +48,7 @@ protected void registerTable(int maxSize)
     @Override
     protected void registerTable()
     {
-        registerTable(SlowQueriesTable.LOGS_VIRTUAL_TABLE_MIN_ROWS);
+        registerTable(1000);
     }
 
     @Test