Queries performed with NODE_LOCAL consistency level do not update req…

…uest metrics

patch by Aleksey Yeschenko; reviewed by Alex Petrov for CASSANDRA-17052

CHANGES.txt

@@ -1,4 +1,5 @@
 4.0.2
+ * Queries performed with NODE_LOCAL consistency level do not update request metrics (CASSANDRA-17052)
  * Fix multiple full sources can be select unexpectedly for bootstrap streaming (CASSANDRA-16945)
  * Fix cassandra.yaml formatting of parameters (CASSANDRA-17131)
  * Add backward compatibility for CQLSSTableWriter Date fields (CASSANDRA-17117)

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

@@ -159,6 +159,11 @@ public enum CassandraRelevantProperties
      */
     REPLACEMENT_ALLOW_EMPTY("cassandra.allow_empty_replace_address", "true"),
 
+    /**
+     * Whether {@link org.apache.cassandra.db.ConsistencyLevel#NODE_LOCAL} should be allowed.
+     */
+    ENABLE_NODELOCAL_QUERIES("cassandra.enable_nodelocal_queries", "false"),
+
     //cassandra properties (without the "cassandra." prefix)
 
     /**

src/java/org/apache/cassandra/cql3/QueryProcessor.java

@@ -37,6 +37,8 @@
 import org.antlr.runtime.*;
 import org.apache.cassandra.concurrent.ScheduledExecutors;
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.metrics.ClientRequestMetrics;
+import org.apache.cassandra.metrics.ClientRequestsMetricsHolder;
 import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.SchemaChangeListener;
 import org.apache.cassandra.schema.SchemaConstants;
@@ -58,6 +60,7 @@
 import org.apache.cassandra.transport.messages.ResultMessage;
 import org.apache.cassandra.utils.*;
 
+import static org.apache.cassandra.config.CassandraRelevantProperties.ENABLE_NODELOCAL_QUERIES;
 import static org.apache.cassandra.cql3.statements.RequestValidations.checkTrue;
 
 public class QueryProcessor implements QueryHandler
@@ -220,19 +223,67 @@ public ResultMessage processStatement(CQLStatement statement, QueryState querySt
         statement.authorize(clientState);
         statement.validate(clientState);
 
-        ResultMessage result;
-        if (options.getConsistency() == ConsistencyLevel.NODE_LOCAL)
+        ResultMessage result = options.getConsistency() == ConsistencyLevel.NODE_LOCAL
+                             ? processNodeLocalStatement(statement, queryState, options)
+                             : statement.execute(queryState, options, queryStartNanoTime);
+
+        return result == null ? new ResultMessage.Void() : result;
+    }
+
+    private ResultMessage processNodeLocalStatement(CQLStatement statement, QueryState queryState, QueryOptions options)
+    {
+        if (!ENABLE_NODELOCAL_QUERIES.getBoolean())
+            throw new InvalidRequestException("NODE_LOCAL consistency level is highly dangerous and should be used only for debugging purposes");
+
+        if (statement instanceof BatchStatement || statement instanceof ModificationStatement)
+            return processNodeLocalWrite(statement, queryState, options);
+        else if (statement instanceof SelectStatement)
+            return processNodeLocalSelect((SelectStatement) statement, queryState, options);
+        else
+            throw new InvalidRequestException("NODE_LOCAL consistency level can only be used with BATCH, UPDATE, INSERT, DELETE, and SELECT statements");
+    }
+
+    private ResultMessage processNodeLocalWrite(CQLStatement statement, QueryState queryState, QueryOptions options)
+    {
+        ClientRequestMetrics  levelMetrics = ClientRequestsMetricsHolder.writeMetricsForLevel(ConsistencyLevel.NODE_LOCAL);
+        ClientRequestMetrics globalMetrics = ClientRequestsMetricsHolder.writeMetrics;
+
+        long startTime = System.nanoTime();
+        try
         {
-            assert Boolean.getBoolean("cassandra.enable_nodelocal_queries") : "Node local consistency level is highly dangerous and should be used only for debugging purposes";
-            assert statement instanceof SelectStatement : "Only SELECT statements are permitted for node-local execution";
-            logger.info("Statement {} executed with NODE_LOCAL consistency level.", statement);
-            result = statement.executeLocally(queryState, options);
+            return statement.executeLocally(queryState, options);
         }
-        else
+        finally
         {
-            result = statement.execute(queryState, options, queryStartNanoTime);
+            long latency = System.nanoTime() - startTime;
+             levelMetrics.addNano(latency);
+            globalMetrics.addNano(latency);
+        }
+    }
+
+    private ResultMessage processNodeLocalSelect(SelectStatement statement, QueryState queryState, QueryOptions options)
+    {
+        ClientRequestMetrics  levelMetrics = ClientRequestsMetricsHolder.readMetricsForLevel(ConsistencyLevel.NODE_LOCAL);
+        ClientRequestMetrics globalMetrics = ClientRequestsMetricsHolder.readMetrics;
+
+        if (StorageService.instance.isBootstrapMode() && !SchemaConstants.isLocalSystemKeyspace(statement.keyspace()))
+        {
+            levelMetrics.unavailables.mark();
+            globalMetrics.unavailables.mark();
+            throw new IsBootstrappingException();
+        }
+
+        long startTime = System.nanoTime();
+        try
+        {
+            return statement.executeLocally(queryState, options);
+        }
+        finally
+        {
+            long latency = System.nanoTime() - startTime;
+             levelMetrics.addNano(latency);
+            globalMetrics.addNano(latency);
         }
-        return result == null ? new ResultMessage.Void() : result;
     }
 
     public static ResultMessage process(String queryString, ConsistencyLevel cl, QueryState queryState, long queryStartNanoTime)

src/java/org/apache/cassandra/metrics/ClientRequestsMetricsHolder.java

@@ -0,0 +1,54 @@
+/*
+ * 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.metrics;
+
+import java.util.EnumMap;
+import java.util.Map;
+
+import org.apache.cassandra.db.ConsistencyLevel;
+
+public final class ClientRequestsMetricsHolder
+{
+    public static final ClientRequestMetrics readMetrics = new ClientRequestMetrics("Read");
+    public static final ClientWriteRequestMetrics writeMetrics = new ClientWriteRequestMetrics("Write");
+    public static final CASClientWriteRequestMetrics casWriteMetrics = new CASClientWriteRequestMetrics("CASWrite");
+    public static final CASClientRequestMetrics casReadMetrics = new CASClientRequestMetrics("CASRead");
+    public static final ViewWriteMetrics viewWriteMetrics = new ViewWriteMetrics("ViewWrite");
+
+    private static final Map<ConsistencyLevel, ClientRequestMetrics> readMetricsMap = new EnumMap<>(ConsistencyLevel.class);
+    private static final Map<ConsistencyLevel, ClientWriteRequestMetrics> writeMetricsMap = new EnumMap<>(ConsistencyLevel.class);
+
+    static
+    {
+        for (ConsistencyLevel level : ConsistencyLevel.values())
+        {
+             readMetricsMap.put(level, new ClientRequestMetrics("Read-" + level.name()));
+            writeMetricsMap.put(level, new ClientWriteRequestMetrics("Write-" + level.name()));
+        }
+    }
+
+    public static ClientRequestMetrics readMetricsForLevel(ConsistencyLevel level)
+    {
+        return readMetricsMap.get(level);
+    }
+
+    public static ClientWriteRequestMetrics writeMetricsForLevel(ConsistencyLevel level)
+    {
+        return writeMetricsMap.get(level);
+    }
+}