Support a means of logging all queries as they were invoked. Patch by Ariel Weisberg; Reviewed by Blake Eggleston
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/ae837806 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/ae837806 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/ae837806 Branch: refs/heads/trunk Commit: ae837806bd07dbb8b881960feeeeb90c1a665d93 Parents: db81f6b Author: Ariel Weisberg <[email protected]> Authored: Fri Oct 27 17:16:45 2017 -0400 Committer: Ariel Weisberg <[email protected]> Committed: Mon Dec 4 18:06:30 2017 -0500 ---------------------------------------------------------------------- CHANGES.txt | 1 + NEWS.txt | 8 + bin/fqltool | 88 +++ bin/fqltool.bat | 36 + build.xml | 17 + conf/cassandra.yaml | 9 +- conf/jvm.options | 4 + lib/chronicle-bytes-1.10.1.jar | Bin 0 -> 273664 bytes lib/chronicle-core-1.9.21.jar | Bin 0 -> 199833 bytes lib/chronicle-queue-4.6.55.jar | Bin 0 -> 215247 bytes lib/chronicle-threads-1.9.1.jar | Bin 0 -> 40530 bytes lib/chronicle-wire-1.10.1.jar | Bin 0 -> 419054 bytes lib/licenses/chronicle-bytes-1.10.1.txt | 14 + lib/licenses/chronicle-core-1.9.21.txt | 14 + lib/licenses/chronicle-queue-4.6.55.txt | 14 + lib/licenses/chronicle-threads-1.9.1.txt | 14 + lib/licenses/chronicle-wire-1.10.1.txt | 14 + .../org/apache/cassandra/config/Config.java | 2 + .../cassandra/config/DatabaseDescriptor.java | 5 + .../db/fullquerylog/FullQueryLogger.java | 530 +++++++++++++++ .../apache/cassandra/service/StorageProxy.java | 23 + .../cassandra/service/StorageProxyMBean.java | 21 + .../cassandra/tools/FullQueryLogTool.java | 93 +++ .../org/apache/cassandra/tools/NodeTool.java | 3 + .../apache/cassandra/tools/fqltool/Dump.java | 265 ++++++++ .../tools/nodetool/DisableFullQueryLog.java | 33 + .../tools/nodetool/EnableFullQueryLog.java | 49 ++ .../tools/nodetool/ResetFullQueryLog.java | 33 + .../transport/messages/BatchMessage.java | 21 + .../transport/messages/ExecuteMessage.java | 11 + .../transport/messages/QueryMessage.java | 11 + .../org/apache/cassandra/utils/ObjectSizes.java | 1 + .../apache/cassandra/utils/binlog/BinLog.java | 277 ++++++++ .../utils/concurrent/WeightedQueue.java | 333 ++++++++++ test/unit/org/apache/cassandra/Util.java | 6 +- .../db/fullquerylog/FullQueryLoggerTest.java | 601 +++++++++++++++++ .../cassandra/utils/binlog/BinLogTest.java | 449 +++++++++++++ .../utils/concurrent/WeightedQueueTest.java | 656 +++++++++++++++++++ 38 files changed, 3652 insertions(+), 4 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 8350848..81da11d 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0 + * Support a means of logging all queries as they were invoked (CASSANDRA-13983) * Presize collections (CASSANDRA-13760) * Add GroupCommitLogService (CASSANDRA-13530) * Parallelize initial materialized view build (CASSANDRA-12245) http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/NEWS.txt ---------------------------------------------------------------------- diff --git a/NEWS.txt b/NEWS.txt index 622cc54..a14f7ba 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -34,6 +34,14 @@ New features threads is specified by the property `cassandra.yaml:concurrent_materialized_view_builders`. This property can be modified at runtime through both JMX and the new `setconcurrentviewbuilders` and `getconcurrentviewbuilders` nodetool commands. See CASSANDRA-12245 for more details. + - There is now a binary full query log based on Chronicle Queue that can be controlled using + nodetool enablefullquerylog, disablefullquerylog, and resetfullquerylog. The log + contains all queries invoked, approximate time they were invoked, any parameters necessary + to bind wildcard values, and all query options. A human readable version of the log can be + dumped or tailed using the new bin/fqltool utility. The full query log is designed to be safe + to use in production and limits utilization of heap memory and disk space with limits + you can specify when enabling the log. + See nodetool and fqltool help text for more information. Upgrading --------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/bin/fqltool ---------------------------------------------------------------------- diff --git a/bin/fqltool b/bin/fqltool new file mode 100755 index 0000000..8a05af1 --- /dev/null +++ b/bin/fqltool @@ -0,0 +1,88 @@ +#!/bin/sh + +# 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. + +if [ "x$CASSANDRA_INCLUDE" = "x" ]; then + # Locations (in order) to use when searching for an include file. + for include in "`dirname "$0"`/cassandra.in.sh" \ + "$HOME/.cassandra.in.sh" \ + /usr/share/cassandra/cassandra.in.sh \ + /usr/local/share/cassandra/cassandra.in.sh \ + /opt/cassandra/cassandra.in.sh; do + if [ -r "$include" ]; then + . "$include" + break + fi + done +elif [ -r "$CASSANDRA_INCLUDE" ]; then + . "$CASSANDRA_INCLUDE" +fi + +# Use JAVA_HOME if set, otherwise look for java in PATH +if [ -x "$JAVA_HOME/bin/java" ]; then + JAVA="$JAVA_HOME/bin/java" +else + JAVA="`which java`" +fi + +if [ "x$JAVA" = "x" ]; then + echo "Java executable not found (hint: set JAVA_HOME)" >&2 + exit 1 +fi + +if [ -z "$CASSANDRA_CONF" -o -z "$CLASSPATH" ]; then + echo "You must set the CASSANDRA_CONF and CLASSPATH vars" >&2 + exit 1 +fi + +# Run cassandra-env.sh to pick up JMX_PORT +if [ -f "$CASSANDRA_CONF/cassandra-env.sh" ]; then + JVM_OPTS_SAVE=$JVM_OPTS + MAX_HEAP_SIZE_SAVE=$MAX_HEAP_SIZE + . "$CASSANDRA_CONF/cassandra-env.sh" + MAX_HEAP_SIZE=$MAX_HEAP_SIZE_SAVE + JVM_OPTS=$JVM_OPTS_SAVE +fi + +# JMX Port passed via cmd line args (-p 9999 / --port 9999 / --port=9999) +# should override the value from cassandra-env.sh +ARGS="" +JVM_ARGS="" +while true +do + if [ ! $1 ]; then break; fi + case $1 in + -D*) + JVM_ARGS="$JVM_ARGS $1" + ;; + *) + ARGS="$ARGS $1" + ;; + esac + shift +done + +if [ "x$MAX_HEAP_SIZE" = "x" ]; then + MAX_HEAP_SIZE="512m" +fi + +"$JAVA" $JAVA_AGENT -ea -da:net.openhft... -cp "$CLASSPATH" $JVM_OPTS -Xmx$MAX_HEAP_SIZE \ + -Dlog4j.configurationFile=log4j2-tools.xml \ + $JVM_ARGS \ + org.apache.cassandra.tools.FullQueryLogTool $ARGS + +# vi:ai sw=4 ts=4 tw=0 et http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/bin/fqltool.bat ---------------------------------------------------------------------- diff --git a/bin/fqltool.bat b/bin/fqltool.bat new file mode 100644 index 0000000..f3103d8 --- /dev/null +++ b/bin/fqltool.bat @@ -0,0 +1,36 @@ +@REM +@REM Licensed to the Apache Software Foundation (ASF) under one or more +@REM contributor license agreements. See the NOTICE file distributed with +@REM this work for additional information regarding copyright ownership. +@REM The ASF licenses this file to You under the Apache License, Version 2.0 +@REM (the "License"); you may not use this file except in compliance with +@REM the License. You may obtain a copy of the License at +@REM +@REM http://www.apache.org/licenses/LICENSE-2.0 +@REM +@REM Unless required by applicable law or agreed to in writing, software +@REM distributed under the License is distributed on an "AS IS" BASIS, +@REM WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +@REM See the License for the specific language governing permissions and +@REM limitations under the License. + +@echo off +if "%OS%" == "Windows_NT" setlocal + +pushd "%~dp0" +call cassandra.in.bat + +if NOT DEFINED JAVA_HOME goto :err + +set CASSANDRA_PARAMS=%CASSANDRA_PARAMS% -Dcassandra.logdir="%CASSANDRA_HOME%\logs" + +"%JAVA_HOME%\bin\java" -cp %CASSANDRA_CLASSPATH% %CASSANDRA_PARAMS% -Dlog4j.configurationFile=log4j2-tools.xml org.apache.cassandra.tools.FullQueryLogTool %* +goto finally + +:err +echo The JAVA_HOME environment variable must be set to run this program! +pause + +:finally +ENDLOCAL & set RC=%ERRORLEVEL% +exit /B %RC% http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/build.xml ---------------------------------------------------------------------- diff --git a/build.xml b/build.xml index 7c63c82..5dee75d 100644 --- a/build.xml +++ b/build.xml @@ -114,6 +114,13 @@ <property name="ecj.version" value="4.4.2"/> + <!-- https://mvnrepository.com/artifact/net.openhft/chronicle-bom/1.15.6 --> + <property name="chronicle-queue.version" value="4.6.55" /> + <property name="chronicle-core.version" value="1.9.21" /> + <property name="chronicle-bytes.version" value="1.10.1" /> + <property name="chronicle-wire.version" value="1.10.1" /> + <property name="chronicle-threads.version" value="1.9.1" /> + <condition property="maven-ant-tasks.jar.exists"> <available file="${build.dir}/maven-ant-tasks-${maven-ant-tasks.version}.jar" /> </condition> @@ -422,6 +429,11 @@ <dependency groupId="org.mindrot" artifactId="jbcrypt" version="0.3m" /> <dependency groupId="io.airlift" artifactId="airline" version="0.8" /> <dependency groupId="io.netty" artifactId="netty-all" version="4.1.14.Final" /> + <dependency groupId="net.openhft" artifactId="chronicle-queue" version="${chronicle-queue.version}"/> + <dependency groupId="net.openhft" artifactId="chronicle-core" version="${chronicle-core.version}"/> + <dependency groupId="net.openhft" artifactId="chronicle-bytes" version="${chronicle-bytes.version}"/> + <dependency groupId="net.openhft" artifactId="chronicle-wire" version="${chronicle-wire.version}"/> + <dependency groupId="net.openhft" artifactId="chronicle-threads" version="${chronicle-threads.version}"/> <dependency groupId="com.google.code.findbugs" artifactId="jsr305" version="2.0.2" /> <dependency groupId="com.clearspring.analytics" artifactId="stream" version="2.5.2" /> <!-- UPDATE AND UNCOMMENT ON THE DRIVER RELEASE, BEFORE 4.0 RELEASE @@ -620,6 +632,11 @@ <dependency groupId="com.github.jbellis" artifactId="jamm"/> <dependency groupId="io.netty" artifactId="netty-all"/> + <dependency groupId="net.openhft" artifactId="chronicle-queue" version="${chronicle-queue.version}"/> + <dependency groupId="net.openhft" artifactId="chronicle-core" version="${chronicle-core.version}"/> + <dependency groupId="net.openhft" artifactId="chronicle-bytes" version="${chronicle-bytes.version}"/> + <dependency groupId="net.openhft" artifactId="chronicle-wire" version="${chronicle-wire.version}"/> + <dependency groupId="net.openhft" artifactId="chronicle-threads" version="${chronicle-threads.version}"/> <dependency groupId="joda-time" artifactId="joda-time"/> <dependency groupId="org.fusesource" artifactId="sigar"/> <dependency groupId="org.eclipse.jdt.core.compiler" artifactId="ecj"/> http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/conf/cassandra.yaml ---------------------------------------------------------------------- diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 9cfc21c..ba478e7 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -1051,9 +1051,9 @@ transparent_data_encryption_options: key_alias: testing:1 # CBC IV length for AES needs to be 16 bytes (which is also the default size) # iv_length: 16 - key_provider: + key_provider: - class_name: org.apache.cassandra.security.JKSKeyProvider - parameters: + parameters: - keystore: conf/.keystore keystore_password: cassandra store_type: JCEKS @@ -1165,3 +1165,8 @@ back_pressure_strategy: # level for writes without timing out. This is different from the consistency level requested by # each write which may be lower in order to facilitate availability. # ideal_consistency_level: EACH_QUORUM + +# Path to write full query log data to when the full query log is enabled +# The full query log will recrusively delete the contents of this path at +# times. Don't place links in this directory to other parts of the filesystem. +#full_query_log_dir: /tmp/cassandrafullquerylog http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/conf/jvm.options ---------------------------------------------------------------------- diff --git a/conf/jvm.options b/conf/jvm.options index d937b08..e875a15 100644 --- a/conf/jvm.options +++ b/conf/jvm.options @@ -86,6 +86,10 @@ # enable assertions. highly suggested for correct application functionality. -ea +# disable assertions for net.openhft.** because it runs out of memory by design +# if enabled and run for more than just brief testing +-da:net.openhft... + # enable thread priorities, primarily so we can give periodic tasks # a lower priority to avoid interfering with client workload -XX:+UseThreadPriorities http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/lib/chronicle-bytes-1.10.1.jar ---------------------------------------------------------------------- diff --git a/lib/chronicle-bytes-1.10.1.jar b/lib/chronicle-bytes-1.10.1.jar new file mode 100644 index 0000000..94645b7 Binary files /dev/null and b/lib/chronicle-bytes-1.10.1.jar differ http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/lib/chronicle-core-1.9.21.jar ---------------------------------------------------------------------- diff --git a/lib/chronicle-core-1.9.21.jar b/lib/chronicle-core-1.9.21.jar new file mode 100644 index 0000000..03436d5 Binary files /dev/null and b/lib/chronicle-core-1.9.21.jar differ http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/lib/chronicle-queue-4.6.55.jar ---------------------------------------------------------------------- diff --git a/lib/chronicle-queue-4.6.55.jar b/lib/chronicle-queue-4.6.55.jar new file mode 100644 index 0000000..09e2b33 Binary files /dev/null and b/lib/chronicle-queue-4.6.55.jar differ http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/lib/chronicle-threads-1.9.1.jar ---------------------------------------------------------------------- diff --git a/lib/chronicle-threads-1.9.1.jar b/lib/chronicle-threads-1.9.1.jar new file mode 100644 index 0000000..5e69515 Binary files /dev/null and b/lib/chronicle-threads-1.9.1.jar differ http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/lib/chronicle-wire-1.10.1.jar ---------------------------------------------------------------------- diff --git a/lib/chronicle-wire-1.10.1.jar b/lib/chronicle-wire-1.10.1.jar new file mode 100644 index 0000000..e6d8b33 Binary files /dev/null and b/lib/chronicle-wire-1.10.1.jar differ http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/lib/licenses/chronicle-bytes-1.10.1.txt ---------------------------------------------------------------------- diff --git a/lib/licenses/chronicle-bytes-1.10.1.txt b/lib/licenses/chronicle-bytes-1.10.1.txt new file mode 100644 index 0000000..d8a262e --- /dev/null +++ b/lib/licenses/chronicle-bytes-1.10.1.txt @@ -0,0 +1,14 @@ + +== Copyright 2016 higherfrequencytrading.com + +Licensed 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. http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/lib/licenses/chronicle-core-1.9.21.txt ---------------------------------------------------------------------- diff --git a/lib/licenses/chronicle-core-1.9.21.txt b/lib/licenses/chronicle-core-1.9.21.txt new file mode 100644 index 0000000..d8a262e --- /dev/null +++ b/lib/licenses/chronicle-core-1.9.21.txt @@ -0,0 +1,14 @@ + +== Copyright 2016 higherfrequencytrading.com + +Licensed 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. http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/lib/licenses/chronicle-queue-4.6.55.txt ---------------------------------------------------------------------- diff --git a/lib/licenses/chronicle-queue-4.6.55.txt b/lib/licenses/chronicle-queue-4.6.55.txt new file mode 100644 index 0000000..d8a262e --- /dev/null +++ b/lib/licenses/chronicle-queue-4.6.55.txt @@ -0,0 +1,14 @@ + +== Copyright 2016 higherfrequencytrading.com + +Licensed 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. http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/lib/licenses/chronicle-threads-1.9.1.txt ---------------------------------------------------------------------- diff --git a/lib/licenses/chronicle-threads-1.9.1.txt b/lib/licenses/chronicle-threads-1.9.1.txt new file mode 100644 index 0000000..d8a262e --- /dev/null +++ b/lib/licenses/chronicle-threads-1.9.1.txt @@ -0,0 +1,14 @@ + +== Copyright 2016 higherfrequencytrading.com + +Licensed 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. http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/lib/licenses/chronicle-wire-1.10.1.txt ---------------------------------------------------------------------- diff --git a/lib/licenses/chronicle-wire-1.10.1.txt b/lib/licenses/chronicle-wire-1.10.1.txt new file mode 100644 index 0000000..d8a262e --- /dev/null +++ b/lib/licenses/chronicle-wire-1.10.1.txt @@ -0,0 +1,14 @@ + +== Copyright 2016 higherfrequencytrading.com + +Licensed 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. http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/src/java/org/apache/cassandra/config/Config.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index bc91df9..1db8217 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -364,6 +364,8 @@ public class Config public RepairCommandPoolFullStrategy repair_command_pool_full_strategy = RepairCommandPoolFullStrategy.queue; public int repair_command_pool_size = concurrent_validations; + public String full_query_log_dir = null; + /** * @deprecated migrate to {@link DatabaseDescriptor#isClientInitialized()} */ http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/src/java/org/apache/cassandra/config/DatabaseDescriptor.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 9083550..843bca2 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -2474,4 +2474,9 @@ public class DatabaseDescriptor { return conf.repair_command_pool_full_strategy; } + + public static String getFullQueryLogPath() + { + return conf.full_query_log_dir; + } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/src/java/org/apache/cassandra/db/fullquerylog/FullQueryLogger.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/fullquerylog/FullQueryLogger.java b/src/java/org/apache/cassandra/db/fullquerylog/FullQueryLogger.java new file mode 100644 index 0000000..d7cf0dc --- /dev/null +++ b/src/java/org/apache/cassandra/db/fullquerylog/FullQueryLogger.java @@ -0,0 +1,530 @@ +/* + * 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.fullquerylog; + +import java.io.File; +import java.nio.ByteBuffer; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +import com.google.common.base.Preconditions; +import com.google.common.collect.Sets; +import com.google.common.primitives.Ints; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.netty.buffer.ByteBuf; +import net.openhft.chronicle.bytes.BytesStore; +import net.openhft.chronicle.queue.RollCycles; +import net.openhft.chronicle.wire.ValueOut; +import net.openhft.chronicle.wire.WireOut; +import org.apache.cassandra.cql3.QueryOptions; +import org.apache.cassandra.io.FSError; +import org.apache.cassandra.io.util.FileUtils; +import org.apache.cassandra.transport.CBUtil; +import org.apache.cassandra.transport.ProtocolVersion; +import org.apache.cassandra.utils.NoSpamLogger; +import org.apache.cassandra.utils.ObjectSizes; +import org.apache.cassandra.utils.Throwables; +import org.apache.cassandra.utils.binlog.BinLog; +import org.apache.cassandra.utils.concurrent.WeightedQueue; +import org.github.jamm.MemoryLayoutSpecification; + +/** + * A logger that logs entire query contents after the query finishes (or times out). + */ +public class FullQueryLogger +{ + private static final int EMPTY_BYTEBUFFER_SIZE = Ints.checkedCast(ObjectSizes.sizeOnHeapExcludingData(ByteBuffer.allocate(0))); + private static final int EMPTY_LIST_SIZE = Ints.checkedCast(ObjectSizes.measureDeep(new ArrayList(0))); + private static final int EMPTY_BYTEBUF_SIZE; + private static final int OBJECT_HEADER_SIZE = MemoryLayoutSpecification.SPEC.getObjectHeaderSize(); + static + { + int tempSize = 0; + ByteBuf buf = CBUtil.allocator.buffer(0, 0); + try + { + tempSize = Ints.checkedCast(ObjectSizes.measure(buf)); + } + finally + { + buf.release(); + } + EMPTY_BYTEBUF_SIZE = tempSize; + } + + private static final Logger logger = LoggerFactory.getLogger(FullQueryLogger.class); + private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1, TimeUnit.MINUTES); + private static final NoSpamLogger.NoSpamLogStatement droppedSamplesStatement = noSpamLogger.getStatement("Dropped {} binary log samples", 1, TimeUnit.MINUTES); + + public static final FullQueryLogger instance = new FullQueryLogger(); + + volatile BinLog binLog; + private volatile boolean blocking; + private Path path; + + private final AtomicLong droppedSamplesSinceLastLog = new AtomicLong(); + + private FullQueryLogger() + { + } + + /** + * Configure the global instance of the FullQueryLogger + * @param path Dedicated path where the FQL can store it's files. + * @param rollCycle How often to roll FQL log segments so they can potentially be reclaimed + * @param blocking Whether the FQL should block if the FQL falls behind or should drop log records + * @param maxQueueWeight Maximum weight of in memory queue for records waiting to be written to the file before blocking or dropping + * @param maxLogSize Maximum size of the rolled files to retain on disk before deleting the oldest file + */ + public synchronized void configure(Path path, String rollCycle, boolean blocking, int maxQueueWeight, long maxLogSize) + { + Preconditions.checkNotNull(path, "path was null"); + File pathAsFile = path.toFile(); + Preconditions.checkNotNull(rollCycle, "rollCycle was null"); + rollCycle = rollCycle.toUpperCase(); + + //Exists and is a directory or can be created + Preconditions.checkArgument((pathAsFile.exists() && pathAsFile.isDirectory()) || (!pathAsFile.exists() && pathAsFile.mkdirs()), "path exists and is not a directory or couldn't be created"); + Preconditions.checkArgument(pathAsFile.canRead() && pathAsFile.canWrite() && pathAsFile.canExecute(), "path is not readable, writable, and executable"); + Preconditions.checkNotNull(RollCycles.valueOf(rollCycle), "unrecognized roll cycle"); + Preconditions.checkArgument(maxQueueWeight > 0, "maxQueueWeight must be > 0"); + Preconditions.checkArgument(maxLogSize > 0, "maxLogSize must be > 0"); + logger.info("Attempting to configure full query logger path: {} Roll cycle: {} Blocking: {} Max queue weight: {} Max log size:{}", path, rollCycle, blocking, maxQueueWeight, maxLogSize); + if (binLog != null) + { + logger.warn("Full query logger already configured. Ignoring requested configuration."); + throw new IllegalStateException("Already configured"); + } + + if (path.toFile().exists()) + { + Throwable error = cleanDirectory(path.toFile(), null); + if (error != null) + { + throw new RuntimeException(error); + } + } + + this.path = path; + this.blocking = blocking; + binLog = new BinLog(path, RollCycles.valueOf(rollCycle), maxQueueWeight, maxLogSize); + binLog.start(); + } + + /** + * Need the path as a parameter as well because if the process is restarted the config file might be the only + * location for retrieving the path to the full query log files, but JMX also allows you to specify a path + * that isn't persisted anywhere so we have to clean that one a well. + */ + public synchronized void reset(String fullQueryLogPath) + { + try + { + Set<File> pathsToClean = Sets.newHashSet(); + + //First decide whether to clean the path configured in the YAML + if (fullQueryLogPath != null) + { + File fullQueryLogPathFile = new File(fullQueryLogPath); + if (fullQueryLogPathFile.exists()) + { + pathsToClean.add(fullQueryLogPathFile); + } + } + + //Then decide whether to clean the last used path, possibly configured by JMX + if (path != null) + { + File pathFile = path.toFile(); + if (pathFile.exists()) + { + pathsToClean.add(pathFile); + } + } + + logger.info("Reset (and deactivation) of full query log requested."); + if (binLog != null) + { + logger.info("Stopping full query log. Cleaning {}.", pathsToClean); + binLog.stop(); + binLog = null; + } + else + { + logger.info("Full query log already deactivated. Cleaning {}.", pathsToClean); + } + + Throwable accumulate = null; + for (File f : pathsToClean) + { + accumulate = cleanDirectory(f, accumulate); + } + if (accumulate != null) + { + throw new RuntimeException(accumulate); + } + } + catch (Exception e) + { + if (e instanceof RuntimeException) + { + throw (RuntimeException)e; + } + throw new RuntimeException(e); + } + } + + /** + * Stop the full query log leaving behind any generated files. + */ + public synchronized void stop() + { + try + { + logger.info("Deactivation of full query log requested."); + if (binLog != null) + { + logger.info("Stopping full query log"); + binLog.stop(); + binLog = null; + } + } + catch (Exception e) + { + throw new RuntimeException(e); + } + } + + /** + * Check whether the full query log is enabled. + * @return true if records are recorded and false otherwise. + */ + public boolean enabled() + { + return binLog != null; + } + + /** + * This is potentially lossy, but it's not super critical as we will always generally know + * when this is happening and roughly how bad it is. + */ + private void logDroppedSample() + { + droppedSamplesSinceLastLog.incrementAndGet(); + if (droppedSamplesStatement.warn(new Object[] {droppedSamplesSinceLastLog.get()})) + { + droppedSamplesSinceLastLog.set(0); + } + } + + /** + * Log an invocation of a batch of queries + * @param type The type of the batch + * @param queries CQL text of the queries + * @param values Values to bind to as parameters for the queries + * @param queryOptions Options associated with the query invocation + * @param batchTimeMillis Approximate time in milliseconds since the epoch since the batch was invoked + */ + public void logBatch(String type, List<String> queries, List<List<ByteBuffer>> values, QueryOptions queryOptions, long batchTimeMillis) + { + Preconditions.checkNotNull(type, "type was null"); + Preconditions.checkNotNull(queries, "queries was null"); + Preconditions.checkNotNull(values, "value was null"); + Preconditions.checkNotNull(queryOptions, "queryOptions was null"); + Preconditions.checkArgument(batchTimeMillis > 0, "batchTimeMillis must be > 0"); + + //Don't construct the wrapper if the log is disabled + BinLog binLog = this.binLog; + if (binLog == null) + { + return; + } + + WeighableMarshallableBatch wrappedBatch = new WeighableMarshallableBatch(type, queries, values, queryOptions, batchTimeMillis); + logRecord(wrappedBatch, binLog); + } + + void logRecord(AbstractWeighableMarshallable record, BinLog binLog) + { + + boolean putInQueue = false; + try + { + if (blocking) + { + try + { + binLog.put(record); + putInQueue = true; + } + catch (InterruptedException e) + { + throw new RuntimeException(e); + } + } + else + { + if (!binLog.offer(record)) + { + logDroppedSample(); + } + else + { + putInQueue = true; + } + } + } + finally + { + if (!putInQueue) + { + record.release(); + } + } + } + + /** + * Log a single CQL query + * @param query CQL query text + * @param queryOptions Options associated with the query invocation + * @param queryTimeMillis Approximate time in milliseconds since the epoch since the batch was invoked + */ + public void logQuery(String query, QueryOptions queryOptions, long queryTimeMillis) + { + Preconditions.checkNotNull(query, "query was null"); + Preconditions.checkNotNull(queryOptions, "queryOptions was null"); + Preconditions.checkArgument(queryTimeMillis > 0, "queryTimeMillis must be > 0"); + + //Don't construct the wrapper if the log is disabled + BinLog binLog = this.binLog; + if (binLog == null) + { + return; + } + + WeighableMarshallableQuery wrappedQuery = new WeighableMarshallableQuery(query, queryOptions, queryTimeMillis); + logRecord(wrappedQuery, binLog); + } + + private static abstract class AbstractWeighableMarshallable extends BinLog.ReleaseableWriteMarshallable implements WeightedQueue.Weighable + { + private final ByteBuf queryOptionsBuffer; + private final long timeMillis; + private final int protocolVersion; + + private AbstractWeighableMarshallable(QueryOptions queryOptions, long timeMillis) + { + this.timeMillis = timeMillis; + ProtocolVersion version = queryOptions.getProtocolVersion(); + this.protocolVersion = version.asInt(); + int optionsSize = QueryOptions.codec.encodedSize(queryOptions, version); + queryOptionsBuffer = CBUtil.allocator.buffer(optionsSize, optionsSize); + /* + * Struggled with what tradeoff to make in terms of query options which is potentially large and complicated + * There is tension between low garbage production (or allocator overhead), small working set size, and CPU overhead reserializing the + * query options into binary format. + * + * I went with the lowest risk most predictable option which is allocator overhead and CPU overhead + * rather then keep the original query message around so I could just serialize that as a memcpy. It's more + * instructions when turned on, but it doesn't change memory footprint quite as much and it's more pay for what you use + * in terms of query volume. The CPU overhead is spread out across producers so we should at least get + * some scaling. + * + */ + boolean success = false; + try + { + QueryOptions.codec.encode(queryOptions, queryOptionsBuffer, version); + success = true; + } + finally + { + if (!success) + { + queryOptionsBuffer.release(); + } + } + } + + @Override + public void writeMarshallable(WireOut wire) + { + wire.write("protocol-version").int32(protocolVersion); + wire.write("query-options").bytes(BytesStore.wrap(queryOptionsBuffer.nioBuffer())); + wire.write("query-time").int64(timeMillis); + } + + @Override + public void release() + { + queryOptionsBuffer.release(); + } + + //8-bytes for protocol version (assume alignment cost), 8-byte timestamp, 8-byte object header + other contents + @Override + public int weight() + { + return 8 + 8 + OBJECT_HEADER_SIZE + EMPTY_BYTEBUF_SIZE + queryOptionsBuffer.capacity(); + } + } + + static class WeighableMarshallableBatch extends AbstractWeighableMarshallable + { + private final int weight; + private final String batchType; + private final List<String> queries; + private final List<List<ByteBuffer>> values; + + public WeighableMarshallableBatch(String batchType, List<String> queries, List<List<ByteBuffer>> values, QueryOptions queryOptions, long batchTimeMillis) + { + super(queryOptions, batchTimeMillis); + this.queries = queries; + this.values = values; + this.batchType = batchType; + boolean success = false; + try + { + + //weight, batch type, queries, values + int weightTemp = 8 + EMPTY_LIST_SIZE + EMPTY_LIST_SIZE; + for (int ii = 0; ii < queries.size(); ii++) + { + weightTemp += ObjectSizes.sizeOf(queries.get(ii)); + } + + weightTemp += EMPTY_LIST_SIZE * values.size(); + for (int ii = 0; ii < values.size(); ii++) + { + List<ByteBuffer> sublist = values.get(ii); + weightTemp += EMPTY_BYTEBUFFER_SIZE * sublist.size(); + for (int zz = 0; zz < sublist.size(); zz++) + { + weightTemp += sublist.get(zz).capacity(); + } + } + weightTemp += super.weight(); + weightTemp += ObjectSizes.sizeOf(batchType); + weight = weightTemp; + success = true; + } + finally + { + if (!success) + { + release(); + } + } + } + + @Override + public void writeMarshallable(WireOut wire) + { + wire.write("type").text("batch"); + super.writeMarshallable(wire); + wire.write("batch-type").text(batchType); + ValueOut valueOut = wire.write("queries"); + valueOut.int32(queries.size()); + for (String query : queries) + { + valueOut.text(query); + } + valueOut = wire.write("values"); + valueOut.int32(values.size()); + for (List<ByteBuffer> subValues : values) + { + valueOut.int32(subValues.size()); + for (ByteBuffer value : subValues) + { + valueOut.bytes(BytesStore.wrap(value)); + } + } + } + + @Override + public int weight() + { + return weight; + } + + } + + static class WeighableMarshallableQuery extends AbstractWeighableMarshallable + { + private final String query; + + public WeighableMarshallableQuery(String query, QueryOptions queryOptions, long queryTimeMillis) + { + super(queryOptions, queryTimeMillis); + this.query = query; + } + + @Override + public void writeMarshallable(WireOut wire) + { + wire.write("type").text("single"); + super.writeMarshallable(wire); + wire.write("query").text(query); + } + + @Override + public int weight() + { + return Ints.checkedCast(ObjectSizes.sizeOf(query)) + super.weight(); + } + } + + static Throwable cleanDirectory(File directory, Throwable accumulate) + { + if (!directory.exists()) + { + return Throwables.merge(accumulate, new RuntimeException(String.format("%s does not exists"))); + } + if (!directory.isDirectory()) + { + return Throwables.merge(accumulate, new RuntimeException(String.format("%s is not a directory", directory))); + } + for (File f : directory.listFiles()) + { + accumulate = deleteRecursively(f, accumulate); + } + if (accumulate instanceof FSError) + { + FileUtils.handleFSError((FSError)accumulate); + } + return accumulate; + } + + private static Throwable deleteRecursively(File fileOrDirectory, Throwable accumulate) + { + if (fileOrDirectory.isDirectory()) + { + for (File f : fileOrDirectory.listFiles()) + { + accumulate = FileUtils.deleteWithConfirm(f, true, accumulate); + } + } + return FileUtils.deleteWithConfirm(fileOrDirectory, true , accumulate); + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/src/java/org/apache/cassandra/service/StorageProxy.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 48d1f3f..aa5d0cc 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.lang.management.ManagementFactory; import java.net.InetAddress; import java.nio.ByteBuffer; +import java.nio.file.Paths; import java.util.*; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; @@ -29,6 +30,7 @@ import java.util.concurrent.atomic.AtomicLong; import javax.management.MBeanServer; import javax.management.ObjectName; +import com.google.common.base.Preconditions; import com.google.common.base.Predicate; import com.google.common.cache.CacheLoader; import com.google.common.collect.*; @@ -51,6 +53,7 @@ import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.db.*; import org.apache.cassandra.db.filter.DataLimits; import org.apache.cassandra.db.filter.TombstoneOverwhelmingException; +import org.apache.cassandra.db.fullquerylog.FullQueryLogger; import org.apache.cassandra.db.partitions.*; import org.apache.cassandra.db.rows.RowIterator; import org.apache.cassandra.db.view.ViewUtils; @@ -2860,6 +2863,26 @@ public class StorageProxy implements StorageProxyMBean return String.format("Updating ideal consistency level new value: %s old value %s", newCL, original.toString()); } + @Override + public void configureFullQueryLogger(String path, String rollCycle, boolean blocking, int maxQueueWeight, long maxLogSize) + { + path = path != null ? path : DatabaseDescriptor.getFullQueryLogPath(); + Preconditions.checkNotNull(path, "cassandra.yaml did not set full_query_log_dir and not set as parameter"); + FullQueryLogger.instance.configure(Paths.get(path), rollCycle, blocking, maxQueueWeight, maxLogSize); + } + + @Override + public void resetFullQueryLogger() + { + FullQueryLogger.instance.reset(DatabaseDescriptor.getFullQueryLogPath()); + } + + @Override + public void stopFullQueryLogger() + { + FullQueryLogger.instance.stop(); + } + public int getOtcBacklogExpirationInterval() { return DatabaseDescriptor.getOtcBacklogExpirationInterval(); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/src/java/org/apache/cassandra/service/StorageProxyMBean.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/StorageProxyMBean.java b/src/java/org/apache/cassandra/service/StorageProxyMBean.java index d271769..173d43f 100644 --- a/src/java/org/apache/cassandra/service/StorageProxyMBean.java +++ b/src/java/org/apache/cassandra/service/StorageProxyMBean.java @@ -71,4 +71,25 @@ public interface StorageProxyMBean public String getIdealConsistencyLevel(); public String setIdealConsistencyLevel(String cl); + + /** + * Start the fully query logger. + * @param path Path where the full query log will be stored. If null cassandra.yaml value is used. + * @param rollCycle How often to create a new file for query data (MINUTELY, DAILY, HOURLY) + * @param blocking Whether threads submitting queries to the query log should block if they can't be drained to the filesystem or alternatively drops samples and log + * @param maxQueueWeight How many bytes of query data to queue before blocking or dropping samples + * @param maxLogSize How many bytes of log data to store before dropping segments. Might not be respected if a log file hasn't rolled so it can be deleted. + */ + public void configureFullQueryLogger(String path, String rollCycle, boolean blocking, int maxQueueWeight, long maxLogSize); + + /** + * Disable the full query logger if it is enabled. + * Also delete any generated files in the last used full query log path as well as the one configure in cassandra.yaml + */ + public void resetFullQueryLogger(); + + /** + * Stop logging queries but leave any generated files on disk. + */ + public void stopFullQueryLogger(); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/src/java/org/apache/cassandra/tools/FullQueryLogTool.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/tools/FullQueryLogTool.java b/src/java/org/apache/cassandra/tools/FullQueryLogTool.java new file mode 100644 index 0000000..0d170d9 --- /dev/null +++ b/src/java/org/apache/cassandra/tools/FullQueryLogTool.java @@ -0,0 +1,93 @@ +/* + * 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.tools; + +import java.util.List; + +import com.google.common.base.Throwables; + +import io.airlift.airline.Cli; +import io.airlift.airline.Help; +import io.airlift.airline.ParseArgumentsMissingException; +import io.airlift.airline.ParseArgumentsUnexpectedException; +import io.airlift.airline.ParseCommandMissingException; +import io.airlift.airline.ParseCommandUnrecognizedException; +import io.airlift.airline.ParseOptionConversionException; +import io.airlift.airline.ParseOptionMissingException; +import io.airlift.airline.ParseOptionMissingValueException; +import org.apache.cassandra.tools.fqltool.Dump; + +import static com.google.common.base.Throwables.getStackTraceAsString; +import static com.google.common.collect.Lists.newArrayList; + +public class FullQueryLogTool +{ + public static void main(String... args) + { + List<Class<? extends Runnable>> commands = newArrayList( + Help.class, + Dump.class + ); + + Cli.CliBuilder<Runnable> builder = Cli.builder("fqltool"); + + builder.withDescription("Manipulate the contents of full query log files") + .withDefaultCommand(Help.class) + .withCommands(commands); + + Cli<Runnable> parser = builder.build(); + + int status = 0; + try + { + parser.parse(args).run(); + } catch (IllegalArgumentException | + IllegalStateException | + ParseArgumentsMissingException | + ParseArgumentsUnexpectedException | + ParseOptionConversionException | + ParseOptionMissingException | + ParseOptionMissingValueException | + ParseCommandMissingException | + ParseCommandUnrecognizedException e) + { + badUse(e); + status = 1; + } catch (Throwable throwable) + { + err(Throwables.getRootCause(throwable)); + status = 2; + } + + System.exit(status); + } + + private static void badUse(Exception e) + { + System.out.println("fqltool: " + e.getMessage()); + System.out.println("See 'fqltool help' or 'fqltool help <command>'."); + } + + private static void err(Throwable e) + { + System.err.println("error: " + e.getMessage()); + System.err.println("-- StackTrace --"); + System.err.println(getStackTraceAsString(e)); + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/src/java/org/apache/cassandra/tools/NodeTool.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/tools/NodeTool.java b/src/java/org/apache/cassandra/tools/NodeTool.java index 0db422e..59d4ead 100644 --- a/src/java/org/apache/cassandra/tools/NodeTool.java +++ b/src/java/org/apache/cassandra/tools/NodeTool.java @@ -76,6 +76,8 @@ public class NodeTool EnableGossip.class, DisableGossip.class, EnableHandoff.class, + EnableFullQueryLog.class, + DisableFullQueryLog.class, GcStats.class, GetBatchlogReplayTrottle.class, GetCompactionThreshold.class, @@ -100,6 +102,7 @@ public class NodeTool Refresh.class, RemoveNode.class, Assassinate.class, + ResetFullQueryLog.class, Repair.class, RepairAdmin.class, ReplayBatchlog.class, http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/src/java/org/apache/cassandra/tools/fqltool/Dump.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/tools/fqltool/Dump.java b/src/java/org/apache/cassandra/tools/fqltool/Dump.java new file mode 100644 index 0000000..9d70b8e --- /dev/null +++ b/src/java/org/apache/cassandra/tools/fqltool/Dump.java @@ -0,0 +1,265 @@ +/* + * 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.tools.fqltool; + +import java.io.File; +import java.nio.BufferUnderflowException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; + +import io.airlift.airline.Arguments; +import io.airlift.airline.Command; +import io.airlift.airline.Option; +import io.netty.buffer.Unpooled; +import net.openhft.chronicle.bytes.Bytes; +import net.openhft.chronicle.queue.ChronicleQueue; +import net.openhft.chronicle.queue.ChronicleQueueBuilder; +import net.openhft.chronicle.queue.ExcerptTailer; +import net.openhft.chronicle.queue.RollCycles; +import net.openhft.chronicle.threads.Pauser; +import net.openhft.chronicle.wire.ReadMarshallable; +import net.openhft.chronicle.wire.ValueIn; +import org.apache.cassandra.cql3.QueryOptions; +import org.apache.cassandra.transport.ProtocolVersion; + +/** + * Dump the contents of a list of paths containing full query logs + */ +@Command(name = "dump", description = "Dump the contents of a full query log") +public class Dump implements Runnable +{ + static final char[] HEXI_DECIMAL = "0123456789ABCDEF".toCharArray(); + + @Arguments(usage = "<path1> [<path2>...<pathN>]", description = "Path containing the full query logs to dump.", required = true) + private List<String> arguments = new ArrayList<>(); + + @Option(title = "roll_cycle", name = {"--roll-cycle"}, description = "How often to roll the log file was rolled. May be necessary for Chronicle to correctly parse file names. (MINUTELY, HOURLY, DAILY). Default HOURLY.") + private String rollCycle = "HOURLY"; + + @Option(title = "follow", name = {"--follow"}, description = "Upon reacahing the end of the log continue indefinitely waiting for more records") + private boolean follow = false; + + @Override + public void run() + { + dump(arguments, rollCycle, follow); + } + + public static void dump(List<String> arguments, String rollCycle, boolean follow) + { + StringBuilder sb = new StringBuilder(); + ReadMarshallable reader = wireIn -> { + sb.setLength(0); + String type = wireIn.read("type").text(); + sb.append("Type: ").append(type).append(System.lineSeparator()); + int protocolVersion = wireIn.read("protocol-version").int32(); + sb.append("Protocol version: ").append(protocolVersion).append(System.lineSeparator()); + QueryOptions options = QueryOptions.codec.decode(Unpooled.wrappedBuffer(wireIn.read("query-options").bytesStore().toTemporaryDirectByteBuffer()), ProtocolVersion.decode(protocolVersion)); + sb.append("Query time: ").append(wireIn.read("query-time").int64()).append(System.lineSeparator()); + if (type.equals("single")) + { + sb.append("Query: ").append(wireIn.read("query").text()).append(System.lineSeparator()); + List<ByteBuffer> values = options.getValues() != null ? options.getValues() : Collections.EMPTY_LIST; + sb.append("Values: ").append(System.lineSeparator()); + valuesToStringBuilder(values, sb); + } + else + { + sb.append("Batch type: ").append(wireIn.read("batch-type").text()).append(System.lineSeparator()); + ValueIn in = wireIn.read("queries"); + int numQueries = in.int32(); + List<String> queries = new ArrayList<>(); + for (int ii = 0; ii < numQueries; ii++) + { + queries.add(in.text()); + } + in = wireIn.read("values"); + int numValues = in.int32(); + List<List<ByteBuffer>> values = new ArrayList<>(); + for (int ii = 0; ii < numValues; ii++) + { + List<ByteBuffer> subValues = new ArrayList<>(); + values.add(subValues); + int numSubValues = in.int32(); + for (int zz = 0; zz < numSubValues; zz++) + { + subValues.add(ByteBuffer.wrap(in.bytes())); + } + sb.append("Query: ").append(queries.get(ii)).append(System.lineSeparator()); + sb.append("Values: ").append(System.lineSeparator()); + valuesToStringBuilder(subValues, sb); + } + } + sb.append(System.lineSeparator()); + System.out.print(sb.toString()); + System.out.flush(); + }; + + //Backoff strategy for spinning on the queue, not aggressive at all as this doesn't need to be low latency + Pauser pauser = Pauser.millis(100); + List<ChronicleQueue> queues = arguments.stream().distinct().map(path -> ChronicleQueueBuilder.single(new File(path)).rollCycle(RollCycles.valueOf(rollCycle)).build()).collect(Collectors.toList()); + List<ExcerptTailer> tailers = queues.stream().map(ChronicleQueue::createTailer).collect(Collectors.toList()); + boolean hadWork = true; + while (hadWork) + { + hadWork = false; + for (ExcerptTailer tailer : tailers) + { + while (tailer.readDocument(reader)) + { + hadWork = true; + } + } + + if (follow) + { + if (!hadWork) + { + //Chronicle queue doesn't support blocking so use this backoff strategy + pauser.pause(); + } + //Don't terminate the loop even if there wasn't work + hadWork = true; + } + } + } + + private static void valuesToStringBuilder(List<ByteBuffer> values, StringBuilder sb) + { + boolean first = true; + for (ByteBuffer value : values) + { + Bytes bytes = Bytes.wrapForRead(value); + long maxLength2 = Math.min(1024, bytes.readLimit() - bytes.readPosition()); + toHexString(bytes, bytes.readPosition(), maxLength2, sb); + if (maxLength2 < bytes.readLimit() - bytes.readPosition()) + { + sb.append("... truncated").append(System.lineSeparator()); + } + + if (first) + { + first = false; + } + else + { + sb.append("-----").append(System.lineSeparator()); + } + } + } + + //This is from net.openhft.chronicle.bytes, need to pass in the StringBuilder so had to copy + /* + * Copyright 2016 higherfrequencytrading.com + * + * Licensed 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. + */ + /** + * display the hex data of {@link Bytes} from the position() to the limit() + * + * @param bytes the buffer you wish to toString() + * @return hex representation of the buffer, from example [0D ,OA, FF] + */ + public static String toHexString(final Bytes bytes, long offset, long len, StringBuilder builder) + throws BufferUnderflowException + { + if (len == 0) + return ""; + + int width = 16; + int[] lastLine = new int[width]; + String sep = ""; + long position = bytes.readPosition(); + long limit = bytes.readLimit(); + + try { + bytes.readPositionRemaining(offset, len); + + long start = offset / width * width; + long end = (offset + len + width - 1) / width * width; + for (long i = start; i < end; i += width) { + // check for duplicate rows + if (i + width < end) { + boolean same = true; + + for (int j = 0; j < width && i + j < offset + len; j++) { + int ch = bytes.readUnsignedByte(i + j); + same &= (ch == lastLine[j]); + lastLine[j] = ch; + } + if (i > start && same) { + sep = "........\n"; + continue; + } + } + builder.append(sep); + sep = ""; + String str = Long.toHexString(i); + for (int j = str.length(); j < 8; j++) + builder.append('0'); + builder.append(str); + for (int j = 0; j < width; j++) { + if (j == width / 2) + builder.append(' '); + if (i + j < offset || i + j >= offset + len) { + builder.append(" "); + + } else { + builder.append(' '); + int ch = bytes.readUnsignedByte(i + j); + builder.append(HEXI_DECIMAL[ch >> 4]); + builder.append(HEXI_DECIMAL[ch & 15]); + } + } + builder.append(' '); + for (int j = 0; j < width; j++) { + if (j == width / 2) + builder.append(' '); + if (i + j < offset || i + j >= offset + len) { + builder.append(' '); + + } else { + int ch = bytes.readUnsignedByte(i + j); + if (ch < ' ' || ch > 126) + ch = '\u00B7'; + builder.append((char) ch); + } + } + builder.append("\n"); + } + return builder.toString(); + } finally { + bytes.readLimit(limit); + bytes.readPosition(position); + } + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/src/java/org/apache/cassandra/tools/nodetool/DisableFullQueryLog.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/tools/nodetool/DisableFullQueryLog.java b/src/java/org/apache/cassandra/tools/nodetool/DisableFullQueryLog.java new file mode 100644 index 0000000..802e854 --- /dev/null +++ b/src/java/org/apache/cassandra/tools/nodetool/DisableFullQueryLog.java @@ -0,0 +1,33 @@ +/* + * 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.tools.nodetool; + +import io.airlift.airline.Command; +import org.apache.cassandra.tools.NodeProbe; +import org.apache.cassandra.tools.NodeTool.NodeToolCmd; + +@Command(name = "disablefullquerylog", description = "Disable the full query log") +public class DisableFullQueryLog extends NodeToolCmd +{ + @Override + public void execute(NodeProbe probe) + { + probe.getSpProxy().stopFullQueryLogger(); + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/src/java/org/apache/cassandra/tools/nodetool/EnableFullQueryLog.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/tools/nodetool/EnableFullQueryLog.java b/src/java/org/apache/cassandra/tools/nodetool/EnableFullQueryLog.java new file mode 100644 index 0000000..624a301 --- /dev/null +++ b/src/java/org/apache/cassandra/tools/nodetool/EnableFullQueryLog.java @@ -0,0 +1,49 @@ +/* + * 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.tools.nodetool; + +import io.airlift.airline.Command; +import io.airlift.airline.Option; +import org.apache.cassandra.tools.NodeProbe; +import org.apache.cassandra.tools.NodeTool.NodeToolCmd; + +@Command(name = "enablefullquerylog", description = "Enable full query logging") +public class EnableFullQueryLog extends NodeToolCmd +{ + @Option(title = "roll_cycle", name = {"--roll-cycle"}, description = "How often to roll the log file (MINUTELY, HOURLY, DAILY). Default HOURLY.") + private String rollCycle = "HOURLY"; + + @Option(title = "blocking", name = {"--blocking"}, description = "If the queue is full whether to block producers or drop samples. Default true.") + private boolean blocking = true; + + @Option(title = "max_queue_weight", name = {"--max-queue-weight"}, description = "Maximum number of bytes of query data to queue to disk before blocking or dropping samples. Default 256 megabytes.") + private int maxQueueWeight = 256 * 1024 * 1024; + + @Option(title = "max_log_size", name = {"--max-log-size"}, description = "How many bytes of log data to store before dropping segments. Might not be respected if a log file hasn't rolled so it can be deleted. Default 16 gigabytes.") + private long maxLogSize = 16L * 1024L * 1024L * 1024L; + + @Option(title = "path", name = {"--path"}, description = "Path to store the full query log at. Will have it's contents recursively deleted. If not set the value from cassandra.yaml will be used.") + private String path = null; + + @Override + public void execute(NodeProbe probe) + { + probe.getSpProxy().configureFullQueryLogger(path, rollCycle, blocking, maxQueueWeight, maxLogSize); + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/src/java/org/apache/cassandra/tools/nodetool/ResetFullQueryLog.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/tools/nodetool/ResetFullQueryLog.java b/src/java/org/apache/cassandra/tools/nodetool/ResetFullQueryLog.java new file mode 100644 index 0000000..cfc20ab --- /dev/null +++ b/src/java/org/apache/cassandra/tools/nodetool/ResetFullQueryLog.java @@ -0,0 +1,33 @@ +/* + * 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.tools.nodetool; + +import io.airlift.airline.Command; +import org.apache.cassandra.tools.NodeProbe; +import org.apache.cassandra.tools.NodeTool.NodeToolCmd; + +@Command(name = "resetfullquerylog", description = "Stop the full query log and clean files in the configured full query log directory from cassandra.yaml as well as JMX") +public class ResetFullQueryLog extends NodeToolCmd +{ + @Override + public void execute(NodeProbe probe) + { + probe.getSpProxy().resetFullQueryLogger(); + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/src/java/org/apache/cassandra/transport/messages/BatchMessage.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/transport/messages/BatchMessage.java b/src/java/org/apache/cassandra/transport/messages/BatchMessage.java index afd7659..dcaa8da 100644 --- a/src/java/org/apache/cassandra/transport/messages/BatchMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/BatchMessage.java @@ -19,6 +19,7 @@ package org.apache.cassandra.transport.messages; import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.UUID; @@ -29,6 +30,7 @@ import org.apache.cassandra.cql3.*; import org.apache.cassandra.cql3.statements.BatchStatement; import org.apache.cassandra.cql3.statements.ModificationStatement; import org.apache.cassandra.cql3.statements.ParsedStatement; +import org.apache.cassandra.db.fullquerylog.FullQueryLogger; import org.apache.cassandra.exceptions.InvalidRequestException; import org.apache.cassandra.exceptions.PreparedQueryNotFoundException; import org.apache.cassandra.service.ClientState; @@ -174,20 +176,30 @@ public class BatchMessage extends Message.Request QueryHandler handler = ClientState.getCQLQueryHandler(); List<ParsedStatement.Prepared> prepared = new ArrayList<>(queryOrIdList.size()); + boolean fullQueryLogEnabled = FullQueryLogger.instance.enabled(); + List<String> queryStrings = fullQueryLogEnabled ? new ArrayList<>(queryOrIdList.size()) : Collections.EMPTY_LIST; for (int i = 0; i < queryOrIdList.size(); i++) { Object query = queryOrIdList.get(i); + String queryString; ParsedStatement.Prepared p; if (query instanceof String) { p = QueryProcessor.parseStatement((String)query, state.getClientState().cloneWithKeyspaceIfSet(options.getKeyspace())); + queryString = (String)query; } else { p = handler.getPrepared((MD5Digest)query); if (p == null) throw new PreparedQueryNotFoundException((MD5Digest)query); + queryString = p.rawCQLStatement; + } + + if (fullQueryLogEnabled) + { + queryStrings.add(queryString); } List<ByteBuffer> queryValues = values.get(i); @@ -215,7 +227,16 @@ public class BatchMessage extends Message.Request // Note: It's ok at this point to pass a bogus value for the number of bound terms in the BatchState ctor // (and no value would be really correct, so we prefer passing a clearly wrong one). BatchStatement batch = new BatchStatement(-1, batchType, statements, Attributes.none()); + long fqlTime = 0; + if (fullQueryLogEnabled) + { + fqlTime = System.currentTimeMillis(); + } Message.Response response = handler.processBatch(batch, state, batchOptions, getCustomPayload(), queryStartNanoTime); + if (fullQueryLogEnabled) + { + FullQueryLogger.instance.logBatch(batchType.name(), queryStrings, values, options, fqlTime); + } if (tracingId != null) response.setTracingId(tracingId); http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java b/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java index 0b93d16..e969134 100644 --- a/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java @@ -31,6 +31,7 @@ import org.apache.cassandra.cql3.statements.BatchStatement; import org.apache.cassandra.cql3.statements.ModificationStatement; import org.apache.cassandra.cql3.statements.ParsedStatement; import org.apache.cassandra.cql3.statements.UpdateStatement; +import org.apache.cassandra.db.fullquerylog.FullQueryLogger; import org.apache.cassandra.exceptions.PreparedQueryNotFoundException; import org.apache.cassandra.service.ClientState; import org.apache.cassandra.service.QueryState; @@ -162,7 +163,17 @@ public class ExecuteMessage extends Message.Request // Some custom QueryHandlers are interested by the bound names. We provide them this information // by wrapping the QueryOptions. QueryOptions queryOptions = QueryOptions.addColumnSpecifications(options, prepared.boundNames); + boolean fqlEnabled = FullQueryLogger.instance.enabled(); + long fqlTime = 0; + if (fqlEnabled) + { + fqlTime = System.currentTimeMillis(); + } Message.Response response = handler.processPrepared(statement, state, queryOptions, getCustomPayload(), queryStartNanoTime); + if (fqlEnabled) + { + FullQueryLogger.instance.logQuery(prepared.rawCQLStatement, options, fqlTime); + } if (response instanceof ResultMessage.Rows) { http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/src/java/org/apache/cassandra/transport/messages/QueryMessage.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/transport/messages/QueryMessage.java b/src/java/org/apache/cassandra/transport/messages/QueryMessage.java index 4c761dd..8f64033 100644 --- a/src/java/org/apache/cassandra/transport/messages/QueryMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/QueryMessage.java @@ -23,6 +23,7 @@ import com.google.common.collect.ImmutableMap; import io.netty.buffer.ByteBuf; import org.apache.cassandra.cql3.QueryOptions; +import org.apache.cassandra.db.fullquerylog.FullQueryLogger; import org.apache.cassandra.exceptions.RequestExecutionException; import org.apache.cassandra.exceptions.RequestValidationException; import org.apache.cassandra.service.ClientState; @@ -113,7 +114,17 @@ public class QueryMessage extends Message.Request Tracing.instance.begin("Execute CQL3 query", state.getClientAddress(), builder.build()); } + boolean fqlEnabled = FullQueryLogger.instance.enabled(); + long fqlTime = 0; + if (fqlEnabled) + { + fqlTime = System.currentTimeMillis(); + } Message.Response response = ClientState.getCQLQueryHandler().process(query, state, options, getCustomPayload(), queryStartNanoTime); + if (fqlEnabled) + { + FullQueryLogger.instance.logQuery(query, options, fqlTime); + } if (options.skipMetadata() && response instanceof ResultMessage.Rows) ((ResultMessage.Rows)response).result.metadata.setSkipMetadata(); http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/src/java/org/apache/cassandra/utils/ObjectSizes.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/utils/ObjectSizes.java b/src/java/org/apache/cassandra/utils/ObjectSizes.java index dea2bac..10dad05 100644 --- a/src/java/org/apache/cassandra/utils/ObjectSizes.java +++ b/src/java/org/apache/cassandra/utils/ObjectSizes.java @@ -138,6 +138,7 @@ public class ObjectSizes * @param str String to calculate memory size of * @return Total in-memory size of the String */ + //@TODO hard coding this to 2 isn't necessarily correct in Java 9 public static long sizeOf(String str) { return STRING_EMPTY_SIZE + sizeOfArray(str.length(), 2); http://git-wip-us.apache.org/repos/asf/cassandra/blob/ae837806/src/java/org/apache/cassandra/utils/binlog/BinLog.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/utils/binlog/BinLog.java b/src/java/org/apache/cassandra/utils/binlog/BinLog.java new file mode 100644 index 0000000..070a151 --- /dev/null +++ b/src/java/org/apache/cassandra/utils/binlog/BinLog.java @@ -0,0 +1,277 @@ +/* + * 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.binlog; + +import java.io.File; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.TimeUnit; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import net.openhft.chronicle.queue.ChronicleQueue; +import net.openhft.chronicle.queue.ChronicleQueueBuilder; +import net.openhft.chronicle.queue.ExcerptAppender; +import net.openhft.chronicle.queue.RollCycle; +import net.openhft.chronicle.queue.impl.StoreFileListener; +import net.openhft.chronicle.wire.WireOut; +import net.openhft.chronicle.wire.WriteMarshallable; +import org.apache.cassandra.concurrent.NamedThreadFactory; +import org.apache.cassandra.utils.concurrent.WeightedQueue; + +/** + * Bin log is a is quick and dirty binary log that is kind of a NIH version of binary logging with a traditional logging + * framework. It's goal is good enough performance, predictable footprint, simplicity in terms of implementation and configuration + * and most importantly minimal impact on producers of log records. + * + * Performance safety is accomplished by feeding items to the binary log using a weighted queue and dropping records if the binary log falls + * sufficiently far behind. + * + * Simplicity and good enough perforamance is achieved by using a single log writing thread as well as Chronicle Queue + * to handle writing the log, making it available for readers, as well as log rolling. + * + */ +public class BinLog implements Runnable, StoreFileListener +{ + private static final Logger logger = LoggerFactory.getLogger(BinLog.class); + + private ChronicleQueue queue; + private ExcerptAppender appender; + @VisibleForTesting + Thread binLogThread = new NamedThreadFactory("Binary Log thread").newThread(this); + final WeightedQueue<ReleaseableWriteMarshallable> sampleQueue; + private final long maxLogSize; + + /** + * The files in the chronicle queue that have already rolled + */ + private Queue<File> chronicleStoreFiles = new ConcurrentLinkedQueue<>(); + + /** + * The number of bytes in store files that have already rolled + */ + private long bytesInStoreFiles; + + private static final ReleaseableWriteMarshallable NO_OP = new ReleaseableWriteMarshallable() + { + @Override + public void writeMarshallable(WireOut wire) + { + } + + @Override + public void release() + { + } + }; + + private volatile boolean shouldContinue = true; + + /** + * + * @param path Path to store the BinLog. Can't be shared with anything else. + * @param rollCycle How often to roll the log file so it can potentially be deleted + * @param maxQueueWeight Maximum weight of in memory queue for records waiting to be written to the file before blocking or dropping + * @param maxLogSize Maximum size of the rolled files to retain on disk before deleting the oldest file + */ + public BinLog(Path path, RollCycle rollCycle, int maxQueueWeight, long maxLogSize) + { + Preconditions.checkNotNull(path, "path was null"); + Preconditions.checkNotNull(rollCycle, "rollCycle was null"); + Preconditions.checkArgument(maxQueueWeight > 0, "maxQueueWeight must be > 0"); + Preconditions.checkArgument(maxLogSize > 0, "maxLogSize must be > 0"); + ChronicleQueueBuilder builder = ChronicleQueueBuilder.single(path.toFile()); + builder.rollCycle(rollCycle); + builder.storeFileListener(this); + queue = builder.build(); + appender = queue.acquireAppender(); + sampleQueue = new WeightedQueue<>(maxQueueWeight); + this.maxLogSize = maxLogSize; + } + + /** + * Start the consumer thread that writes log records. Can only be done once. + */ + public void start() + { + if (!shouldContinue) + { + throw new IllegalStateException("Can't reuse stopped BinLog"); + } + binLogThread.start(); + } + + /** + * Stop the consumer thread that writes log records. Can be called multiple times. + * @throws InterruptedException + */ + public synchronized void stop() throws InterruptedException + { + if (!shouldContinue) + { + return; + } + + shouldContinue = false; + sampleQueue.put(NO_OP); + binLogThread.join(); + appender = null; + queue = null; + } + + /** + * Offer a record to the log. If the in memory queue is full the record will be dropped and offer will return false. + * @param record The record to write to the log + * @return true if the record was queued and false otherwise + */ + public boolean offer(ReleaseableWriteMarshallable record) + { + if (!shouldContinue) + { + return false; + } + + return sampleQueue.offer(record); + } + + /** + * Put a record into the log. If the in memory queue is full the putting thread will be blocked until there is space or it is interrupted. + * @param record The record to write to the log + * @throws InterruptedException + */ + public void put(ReleaseableWriteMarshallable record) throws InterruptedException + { + if (!shouldContinue) + { + return; + } + + //Resolve potential deadlock at shutdown when queue is full + while (shouldContinue) + { + if (sampleQueue.offer(record, 1, TimeUnit.SECONDS)) + { + return; + } + } + } + + private void processTasks(List<ReleaseableWriteMarshallable> tasks) + { + for (int ii = 0; ii < tasks.size(); ii++) + { + WriteMarshallable t = tasks.get(ii); + //Don't write an empty document + if (t == NO_OP) + { + continue; + } + + appender.writeDocument(t); + } + } + + @Override + public void run() + { + List<ReleaseableWriteMarshallable> tasks = new ArrayList<>(16); + while (shouldContinue) + { + try + { + tasks.clear(); + ReleaseableWriteMarshallable task = sampleQueue.take(); + tasks.add(task); + sampleQueue.drainTo(tasks, 15); + + processTasks(tasks); + } + catch (Throwable t) + { + logger.error("Unexpected exception in binary log thread", t); + } + finally + { + for (int ii = 0; ii < tasks.size(); ii++) + { + tasks.get(ii).release(); + } + } + } + + //Clean up the buffers on thread exit, finalization will check again once this + //is no longer reachable ensuring there are no stragglers in the queue. + finalize(); + } + + /** + * Track store files as they are added and their storage impact. Delete them if over storage limit. + * @param cycle + * @param file + */ + public synchronized void onReleased(int cycle, File file) + { + chronicleStoreFiles.offer(file); + //This isn't accurate because the files are sparse, but it's at least pessimistic + bytesInStoreFiles += file.length(); + logger.debug("Chronicle store file {} rolled file size {}", file.getPath(), file.length()); + while (bytesInStoreFiles > maxLogSize & !chronicleStoreFiles.isEmpty()) + { + File toDelete = chronicleStoreFiles.poll(); + long toDeleteLength = toDelete.length(); + if (!toDelete.delete()) + { + logger.error("Failed to delete chronicle store file: {} store file size: {} bytes in store files: {}. " + + "You will need to clean this up manually or reset full query logging.", + toDelete.getPath(), toDeleteLength, bytesInStoreFiles); + } + else + { + bytesInStoreFiles -= toDeleteLength; + logger.info("Deleted chronicle store file: {} store file size: {} bytes in store files: {} max log size: {}.", file.getPath(), toDeleteLength, bytesInStoreFiles, maxLogSize); + } + } + } + + /** + * There is a race where we might not release a buffer, going to let finalization + * catch it since it shouldn't happen to a lot of buffers. Only test code would run + * into it anyways. + */ + @Override + public void finalize() + { + ReleaseableWriteMarshallable toRelease; + while (((toRelease = sampleQueue.poll()) != null)) + { + toRelease.release(); + } + } + + public abstract static class ReleaseableWriteMarshallable implements WriteMarshallable + { + protected abstract void release(); + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
