diff --git a/CHANGES.txt b/CHANGES.txt index 43f8cbeaef..3d1991a4f8 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0-alpha3 + * Align record header of FQL and audit binary log (CASSANDRA-15076) * Shuffle forwarding replica for messages to non-local DC (CASSANDRA-15318) * Optimise native protocol ASCII string encoding (CASSANDRA-15410) * Make sure all exceptions are propagated in DebuggableThreadPoolExecutor (CASSANDRA-15332) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index f3e5c7507c..9a79f24d7e 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -1259,11 +1259,6 @@ back_pressure_strategy: # 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 - # Automatically upgrade sstables after upgrade - if there is no ordinary compaction to do, the # oldest non-upgraded sstable will get upgraded to the latest version # automatic_sstable_upgrade: false diff --git a/src/java/org/apache/cassandra/audit/BinAuditLogger.java b/src/java/org/apache/cassandra/audit/BinAuditLogger.java index 23b9977c46..83ed3dea28 100644 --- a/src/java/org/apache/cassandra/audit/BinAuditLogger.java +++ b/src/java/org/apache/cassandra/audit/BinAuditLogger.java @@ -30,8 +30,8 @@ import org.apache.cassandra.utils.concurrent.WeightedQueue; public class BinAuditLogger extends BinLogAuditLogger implements IAuditLogger { - public static final String TYPE = "type"; - public static final String AUDITLOG_TYPE = "AuditLog"; + public static final long CURRENT_VERSION = 0; + public static final String AUDITLOG_TYPE = "audit"; public static final String AUDITLOG_MESSAGE = "message"; public BinAuditLogger() @@ -71,10 +71,19 @@ public class BinAuditLogger extends BinLogAuditLogger implements IAuditLogger this.message = message; } - @Override - public void writeMarshallable(WireOut wire) + protected long version() + { + return CURRENT_VERSION; + } + + protected String type() + { + return AUDITLOG_TYPE; + } + + @Override + public void writeMarshallablePayload(WireOut wire) { - wire.write(TYPE).text(AUDITLOG_TYPE); wire.write(AUDITLOG_MESSAGE).text(message); } diff --git a/src/java/org/apache/cassandra/audit/FullQueryLogger.java b/src/java/org/apache/cassandra/audit/FullQueryLogger.java index 9c1f472aa3..79829408e9 100644 --- a/src/java/org/apache/cassandra/audit/FullQueryLogger.java +++ b/src/java/org/apache/cassandra/audit/FullQueryLogger.java @@ -23,7 +23,6 @@ import java.util.List; import javax.annotation.Nullable; -import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import com.google.common.primitives.Ints; @@ -45,10 +44,7 @@ import org.github.jamm.MemoryLayoutSpecification; */ public class FullQueryLogger extends BinLogAuditLogger implements IAuditLogger { - public static final long CURRENT_VERSION = 0; // encode a dummy version, to prevent pain in decoding in the future - - public static final String VERSION = "version"; - public static final String TYPE = "type"; + public static final long CURRENT_VERSION = 0; public static final String PROTOCOL_VERSION = "protocol-version"; public static final String QUERY_OPTIONS = "query-options"; @@ -169,9 +165,9 @@ public class FullQueryLogger extends BinLogAuditLogger implements IAuditLogger } @Override - public void writeMarshallable(WireOut wire) + public void writeMarshallablePayload(WireOut wire) { - super.writeMarshallable(wire); + super.writeMarshallablePayload(wire); wire.write(QUERY).text(query); } @@ -229,9 +225,9 @@ public class FullQueryLogger extends BinLogAuditLogger implements IAuditLogger } @Override - public void writeMarshallable(WireOut wire) + public void writeMarshallablePayload(WireOut wire) { - super.writeMarshallable(wire); + super.writeMarshallablePayload(wire); wire.write(BATCH_TYPE).text(batchType.name()); ValueOut valueOut = wire.write(QUERIES); valueOut.int32(queries.size()); @@ -305,11 +301,14 @@ public class FullQueryLogger extends BinLogAuditLogger implements IAuditLogger } @Override - public void writeMarshallable(WireOut wire) + protected long version() { - wire.write(VERSION).int16(CURRENT_VERSION); - wire.write(TYPE).text(type()); + return CURRENT_VERSION; + } + @Override + public void writeMarshallablePayload(WireOut wire) + { wire.write(QUERY_START_TIME).int64(queryStartTime); wire.write(PROTOCOL_VERSION).int32(protocolVersion); wire.write(QUERY_OPTIONS).bytes(BytesStore.wrap(queryOptionsBuffer.nioBuffer())); @@ -339,8 +338,6 @@ public class FullQueryLogger extends BinLogAuditLogger implements IAuditLogger ? Ints.checkedCast(ObjectSizes.sizeOf(keyspace)) // keyspace : OBJECT_REFERENCE_SIZE); // null } - - protected abstract String type(); } } diff --git a/src/java/org/apache/cassandra/tools/AuditLogViewer.java b/src/java/org/apache/cassandra/tools/AuditLogViewer.java index 01ea7b37ca..f1a6e3723a 100644 --- a/src/java/org/apache/cassandra/tools/AuditLogViewer.java +++ b/src/java/org/apache/cassandra/tools/AuditLogViewer.java @@ -40,6 +40,7 @@ import net.openhft.chronicle.threads.Pauser; import net.openhft.chronicle.wire.ReadMarshallable; import net.openhft.chronicle.wire.WireIn; import org.apache.cassandra.audit.BinAuditLogger; +import org.apache.cassandra.utils.binlog.BinLog; /** * Tool to view the contenst of AuditLog files in human readable format. Default implementation for AuditLog files @@ -51,6 +52,7 @@ public class AuditLogViewer private static final String TOOL_NAME = "auditlogviewer"; private static final String ROLL_CYCLE = "roll_cycle"; private static final String FOLLOW = "follow"; + private static final String IGNORE = "ignore"; private static final String HELP_OPTION = "help"; public static void main(String[] args) @@ -59,7 +61,7 @@ public class AuditLogViewer try { - dump(options.pathList, options.rollCycle, options.follow, System.out::print); + dump(options.pathList, options.rollCycle, options.follow, options.ignoreUnsupported, System.out::print); } catch (Exception e) { @@ -68,7 +70,7 @@ public class AuditLogViewer } } - static void dump(List pathList, String rollCycle, boolean follow, Consumer displayFun) + static void dump(List pathList, String rollCycle, boolean follow, boolean ignoreUnsupported, Consumer displayFun) { //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); @@ -83,7 +85,7 @@ public class AuditLogViewer hadWork = false; for (ExcerptTailer tailer : tailers) { - while (tailer.readDocument(new DisplayRecord(displayFun))) + while (tailer.readDocument(new DisplayRecord(ignoreUnsupported, displayFun))) { hadWork = true; } @@ -104,29 +106,70 @@ public class AuditLogViewer private static class DisplayRecord implements ReadMarshallable { + private final boolean ignoreUnsupported; private final Consumer displayFun; - DisplayRecord(Consumer displayFun) + DisplayRecord(boolean ignoreUnsupported, Consumer displayFun) { + this.ignoreUnsupported = ignoreUnsupported; this.displayFun = displayFun; } public void readMarshallable(WireIn wireIn) throws IORuntimeException { - StringBuilder sb = new StringBuilder(); - - String type = wireIn.read(BinAuditLogger.TYPE).text(); - sb.append("Type: ") - .append(type) - .append(System.lineSeparator()); - - if (null != type && type.equals(BinAuditLogger.AUDITLOG_TYPE)) + int version = wireIn.read(BinLog.VERSION).int16(); + if (!isSupportedVersion(version)) { - sb.append("LogMessage: ").append(wireIn.read(BinAuditLogger.AUDITLOG_MESSAGE).text()).append(System.lineSeparator()); + return; + } + String type = wireIn.read(BinLog.TYPE).text(); + if (!isSupportedType(type)) + { + return; } + StringBuilder sb = new StringBuilder(); + sb.append("Type: ") + .append(type) + .append(System.lineSeparator()) + .append("LogMessage: ") + .append(wireIn.read(BinAuditLogger.AUDITLOG_MESSAGE).text()) + .append(System.lineSeparator()); + displayFun.accept(sb.toString()); } + + private boolean isSupportedVersion(int version) + { + if (version <= BinAuditLogger.CURRENT_VERSION) + { + return true; + } + + if (ignoreUnsupported) + { + return false; + } + + throw new IORuntimeException("Unsupported record version [" + version + + "] - highest supported version is [" + BinAuditLogger.CURRENT_VERSION + ']'); + } + + private boolean isSupportedType(String type) + { + if (BinAuditLogger.AUDITLOG_TYPE.equals(type)) + { + return true; + } + + if (ignoreUnsupported) + { + return false; + } + + throw new IORuntimeException("Unsupported record type field [" + type + + "] - supported type is [" + BinAuditLogger.AUDITLOG_TYPE + ']'); + } } private static class AuditLogViewerOptions @@ -134,6 +177,7 @@ public class AuditLogViewer private final List pathList; private String rollCycle = "HOURLY"; private boolean follow; + private boolean ignoreUnsupported; private AuditLogViewerOptions(String[] pathList) { @@ -166,6 +210,8 @@ public class AuditLogViewer opts.follow = cmd.hasOption(FOLLOW); + opts.ignoreUnsupported = cmd.hasOption(IGNORE); + if (cmd.hasOption(ROLL_CYCLE)) { opts.rollCycle = cmd.getOptionValue(ROLL_CYCLE); @@ -191,8 +237,9 @@ public class AuditLogViewer { Options options = new Options(); - options.addOption(new Option("r", ROLL_CYCLE, false, "How often to roll the log file was rolled. May be necessary for Chronicle to correctly parse file names. (MINUTELY, HOURLY, DAILY). Default HOURLY.")); + options.addOption(new Option("r", ROLL_CYCLE, true, "How often to roll the log file was rolled. May be necessary for Chronicle to correctly parse file names. (MINUTELY, HOURLY, DAILY). Default HOURLY.")); options.addOption(new Option("f", FOLLOW, false, "Upon reacahing the end of the log continue indefinitely waiting for more records")); + options.addOption(new Option("i", IGNORE, false, "Silently ignore unsupported records")); options.addOption(new Option("h", HELP_OPTION, false, "display this help message")); return options; diff --git a/src/java/org/apache/cassandra/utils/binlog/BinLog.java b/src/java/org/apache/cassandra/utils/binlog/BinLog.java index d4dac78667..0fce6eec37 100644 --- a/src/java/org/apache/cassandra/utils/binlog/BinLog.java +++ b/src/java/org/apache/cassandra/utils/binlog/BinLog.java @@ -53,6 +53,9 @@ public class BinLog implements Runnable { private static final Logger logger = LoggerFactory.getLogger(BinLog.class); + public static final String VERSION = "version"; + public static final String TYPE = "type"; + private ChronicleQueue queue; private ExcerptAppender appender; @VisibleForTesting @@ -63,7 +66,19 @@ public class BinLog implements Runnable private static final ReleaseableWriteMarshallable NO_OP = new ReleaseableWriteMarshallable() { @Override - public void writeMarshallable(WireOut wire) + protected long version() + { + return 0; + } + + @Override + protected String type() + { + return "no-op"; + } + + @Override + public void writeMarshallablePayload(WireOut wire) { } @@ -229,6 +244,21 @@ public class BinLog implements Runnable public abstract static class ReleaseableWriteMarshallable implements WriteMarshallable { + @Override + public final void writeMarshallable(WireOut wire) + { + wire.write(VERSION).int16(version()); + wire.write(TYPE).text(type()); + + writeMarshallablePayload(wire); + } + + protected abstract long version(); + + protected abstract String type(); + + protected abstract void writeMarshallablePayload(WireOut wire); + public abstract void release(); } } diff --git a/test/unit/org/apache/cassandra/audit/BinAuditLoggerTest.java b/test/unit/org/apache/cassandra/audit/BinAuditLoggerTest.java index f9d2930ea5..90c93250e4 100644 --- a/test/unit/org/apache/cassandra/audit/BinAuditLoggerTest.java +++ b/test/unit/org/apache/cassandra/audit/BinAuditLoggerTest.java @@ -76,12 +76,14 @@ public class BinAuditLoggerTest extends CQLTester { ExcerptTailer tailer = queue.createTailer(); assertTrue(tailer.readDocument(wire -> { - assertEquals("AuditLog", wire.read("type").text()); + assertEquals(0L, wire.read("version").int16()); + assertEquals("audit", wire.read("type").text()); assertThat(wire.read("message").text(), containsString(AuditLogEntryType.PREPARE_STATEMENT.toString())); })); assertTrue(tailer.readDocument(wire -> { - assertEquals("AuditLog", wire.read("type").text()); + assertEquals(0L, wire.read("version").int16()); + assertEquals("audit", wire.read("type").text()); assertThat(wire.read("message").text(), containsString(AuditLogEntryType.SELECT.toString())); })); assertFalse(tailer.readDocument(wire -> { diff --git a/test/unit/org/apache/cassandra/audit/FullQueryLoggerTest.java b/test/unit/org/apache/cassandra/audit/FullQueryLoggerTest.java index 525fa8ef84..62e5a50a1d 100644 --- a/test/unit/org/apache/cassandra/audit/FullQueryLoggerTest.java +++ b/test/unit/org/apache/cassandra/audit/FullQueryLoggerTest.java @@ -55,11 +55,9 @@ import org.apache.cassandra.service.QueryState; import org.apache.cassandra.transport.ProtocolVersion; import org.apache.cassandra.utils.ObjectSizes; import org.apache.cassandra.utils.binlog.BinLogTest; -import org.apache.cassandra.utils.binlog.DeletingArchiver; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.apache.cassandra.audit.FullQueryLogger.BATCH; @@ -72,9 +70,10 @@ import static org.apache.cassandra.audit.FullQueryLogger.QUERY; import static org.apache.cassandra.audit.FullQueryLogger.QUERY_OPTIONS; import static org.apache.cassandra.audit.FullQueryLogger.QUERY_START_TIME; import static org.apache.cassandra.audit.FullQueryLogger.SINGLE_QUERY; -import static org.apache.cassandra.audit.FullQueryLogger.TYPE; import static org.apache.cassandra.audit.FullQueryLogger.VALUES; -import static org.apache.cassandra.audit.FullQueryLogger.VERSION; + +import static org.apache.cassandra.utils.binlog.BinLog.TYPE; +import static org.apache.cassandra.utils.binlog.BinLog.VERSION; public class FullQueryLoggerTest extends CQLTester { @@ -274,7 +273,7 @@ public class FullQueryLoggerTest extends CQLTester instance.binLog.put(new Query("foo1", QueryOptions.DEFAULT, queryState(), 1) { - public void writeMarshallable(WireOut wire) + public void writeMarshallablePayload(WireOut wire) { //Notify that the bin log is blocking now binLogBlocked.release(); @@ -287,7 +286,7 @@ public class FullQueryLoggerTest extends CQLTester { e.printStackTrace(); } - super.writeMarshallable(wire); + super.writeMarshallablePayload(wire); } public void release() @@ -356,7 +355,7 @@ public class FullQueryLoggerTest extends CQLTester instance.binLog.put(new Query("foo1", QueryOptions.DEFAULT, queryState(), 1) { - public void writeMarshallable(WireOut wire) + public void writeMarshallablePayload(WireOut wire) { //Notify that the bin log is blocking now binLogBlocked.release(); @@ -369,7 +368,7 @@ public class FullQueryLoggerTest extends CQLTester { e.printStackTrace(); } - super.writeMarshallable(wire); + super.writeMarshallablePayload(wire); } public void release() @@ -388,10 +387,10 @@ public class FullQueryLoggerTest extends CQLTester AtomicInteger releasedCount = new AtomicInteger(0); AtomicInteger writtenCount = new AtomicInteger(0); instance.logRecord(new Query("foo3", QueryOptions.DEFAULT, queryState(), 1) { - public void writeMarshallable(WireOut wire) + public void writeMarshallablePayload(WireOut wire) { writtenCount.incrementAndGet(); - super.writeMarshallable(wire); + super.writeMarshallablePayload(wire); } public void release() diff --git a/test/unit/org/apache/cassandra/tools/AuditLogViewerTest.java b/test/unit/org/apache/cassandra/tools/AuditLogViewerTest.java index 078f899b88..649712a8e5 100644 --- a/test/unit/org/apache/cassandra/tools/AuditLogViewerTest.java +++ b/test/unit/org/apache/cassandra/tools/AuditLogViewerTest.java @@ -31,12 +31,16 @@ import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import net.openhft.chronicle.core.io.IORuntimeException; import net.openhft.chronicle.queue.ChronicleQueue; import net.openhft.chronicle.queue.ChronicleQueueBuilder; import net.openhft.chronicle.queue.ExcerptAppender; import net.openhft.chronicle.queue.RollCycles; +import net.openhft.chronicle.wire.WireOut; import org.apache.cassandra.audit.BinAuditLogger; +import static org.junit.Assert.assertTrue; + public class AuditLogViewerTest { private Path path; @@ -73,12 +77,138 @@ public class AuditLogViewerTest //Read those written records List actualRecords = new ArrayList<>(); - AuditLogViewer.dump(ImmutableList.of(path.toString()), RollCycles.TEST_SECONDLY.toString(), false, actualRecords::add); + AuditLogViewer.dump(ImmutableList.of(path.toString()), RollCycles.TEST_SECONDLY.toString(), false, false, actualRecords::add); - for (int i = 0; i < records.size(); i++) - { - Assert.assertTrue(actualRecords.get(i).contains(records.get(i))); - } + assertRecordsMatch(records, actualRecords); } } -} \ No newline at end of file + + @Test (expected = IORuntimeException.class) + public void testRejectFutureVersionRecord() + { + try (ChronicleQueue queue = ChronicleQueueBuilder.single(path.toFile()).rollCycle(RollCycles.TEST_SECONDLY).build()) + { + ExcerptAppender appender = queue.acquireAppender(); + appender.writeDocument(createFutureRecord()); + + AuditLogViewer.dump(ImmutableList.of(path.toString()), RollCycles.TEST_SECONDLY.toString(), false, false, dummy -> {}); + } + catch (Exception e) + { + assertTrue(e.getMessage().contains("Unsupported record version")); + throw e; + } + } + + @Test + public void testIgnoreFutureVersionRecord() + { + List records = new ArrayList<>(); + records.add("Test foo bar 1"); + records.add("Test foo bar 2"); + + try (ChronicleQueue queue = ChronicleQueueBuilder.single(path.toFile()).rollCycle(RollCycles.TEST_SECONDLY).build()) + { + ExcerptAppender appender = queue.acquireAppender(); + + //Write future record + appender.writeDocument(createFutureRecord()); + + //Write bunch of current records + records.forEach(s -> appender.writeDocument(new BinAuditLogger.Message(s))); + + //Read those written records + List actualRecords = new ArrayList<>(); + AuditLogViewer.dump(ImmutableList.of(path.toString()), RollCycles.TEST_SECONDLY.toString(), false, true, actualRecords::add); + + // Assert all current records are present + assertRecordsMatch(records, actualRecords); + } + } + + @Test (expected = IORuntimeException.class) + public void testRejectUnknownTypeRecord() + { + try (ChronicleQueue queue = ChronicleQueueBuilder.single(path.toFile()).rollCycle(RollCycles.TEST_SECONDLY).build()) + { + ExcerptAppender appender = queue.acquireAppender(); + appender.writeDocument(createUnknownTypeRecord()); + + AuditLogViewer.dump(ImmutableList.of(path.toString()), RollCycles.TEST_SECONDLY.toString(), false, false, dummy -> {}); + } + catch (Exception e) + { + assertTrue(e.getMessage().contains("Unsupported record type field")); + throw e; + } + } + + @Test + public void testIgnoreUnknownTypeRecord() + { + List records = new ArrayList<>(); + records.add("Test foo bar 1"); + records.add("Test foo bar 2"); + + try (ChronicleQueue queue = ChronicleQueueBuilder.single(path.toFile()).rollCycle(RollCycles.TEST_SECONDLY).build()) + { + ExcerptAppender appender = queue.acquireAppender(); + + //Write unrecognized type record + appender.writeDocument(createUnknownTypeRecord()); + + //Write bunch of supported records + records.forEach(s -> appender.writeDocument(new BinAuditLogger.Message(s))); + + //Read those written records + List actualRecords = new ArrayList<>(); + AuditLogViewer.dump(ImmutableList.of(path.toString()), RollCycles.TEST_SECONDLY.toString(), false, true, actualRecords::add); + + // Assert all supported records are present + assertRecordsMatch(records, actualRecords); + } + } + + private BinAuditLogger.Message createFutureRecord() + { + return new BinAuditLogger.Message("dummy message") { + protected long version() + { + return 999; + } + + @Override + public void writeMarshallablePayload(WireOut wire) + { + super.writeMarshallablePayload(wire); + wire.write("future-field").text("future_value"); + } + }; + } + + private BinAuditLogger.Message createUnknownTypeRecord() + { + return new BinAuditLogger.Message("dummy message") { + protected String type() + { + return "unknown-type"; + } + + @Override + public void writeMarshallablePayload(WireOut wire) + { + super.writeMarshallablePayload(wire); + wire.write("unknown-field").text("unknown_value"); + } + }; + } + + private void assertRecordsMatch(List records, List actualRecords) + { + Assert.assertEquals(records.size(), actualRecords.size()); + for (int i = 0; i < records.size(); i++) + { + Assert.assertTrue(actualRecords.get(i).contains(records.get(i))); + } + } +} diff --git a/test/unit/org/apache/cassandra/utils/binlog/BinLogTest.java b/test/unit/org/apache/cassandra/utils/binlog/BinLogTest.java index 76c42f2739..81f1ea0e42 100644 --- a/test/unit/org/apache/cassandra/utils/binlog/BinLogTest.java +++ b/test/unit/org/apache/cassandra/utils/binlog/BinLogTest.java @@ -36,8 +36,6 @@ import net.openhft.chronicle.queue.ExcerptTailer; import net.openhft.chronicle.queue.RollCycles; import net.openhft.chronicle.wire.WireOut; import org.apache.cassandra.Util; -import org.apache.cassandra.audit.AuditLogOptions; -import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.io.util.FileUtils; import static org.junit.Assert.assertEquals; @@ -122,7 +120,17 @@ public class BinLogTest releaseCount.incrementAndGet(); } - public void writeMarshallable(WireOut wire) + protected long version() + { + return 0; + } + + protected String type() + { + return "test"; + } + + public void writeMarshallablePayload(WireOut wire) { try { @@ -136,8 +144,17 @@ public class BinLogTest }); binLog.put(new BinLog.ReleaseableWriteMarshallable() { + protected long version() + { + return 0; + } - public void writeMarshallable(WireOut wire) + protected String type() + { + return "test"; + } + + public void writeMarshallablePayload(WireOut wire) { } @@ -182,7 +199,17 @@ public class BinLogTest released.release(); } - public void writeMarshallable(WireOut wire) + protected long version() + { + return 0; + } + + protected String type() + { + return "test"; + } + + public void writeMarshallablePayload(WireOut wire) { } @@ -227,7 +254,17 @@ public class BinLogTest { } - public void writeMarshallable(WireOut wire) + protected long version() + { + return 0; + } + + protected String type() + { + return "test"; + } + + public void writeMarshallablePayload(WireOut wire) { //Notify the bing log thread is about to block binLogBlocked.release(); @@ -303,7 +340,17 @@ public class BinLogTest { } - public void writeMarshallable(WireOut wire) + protected long version() + { + return 0; + } + + protected String type() + { + return "test"; + } + + public void writeMarshallablePayload(WireOut wire) { //Notify the bing log thread is about to block binLogBlocked.release(); @@ -420,7 +467,17 @@ public class BinLogTest //Do nothing } - public void writeMarshallable(WireOut wire) + protected long version() + { + return 0; + } + + protected String type() + { + return "test"; + } + + public void writeMarshallablePayload(WireOut wire) { wire.write("text").text(text); } diff --git a/tools/fqltool/src/org/apache/cassandra/fqltool/FQLQueryReader.java b/tools/fqltool/src/org/apache/cassandra/fqltool/FQLQueryReader.java index 0c18b796ed..164b077faa 100644 --- a/tools/fqltool/src/org/apache/cassandra/fqltool/FQLQueryReader.java +++ b/tools/fqltool/src/org/apache/cassandra/fqltool/FQLQueryReader.java @@ -32,14 +32,13 @@ import net.openhft.chronicle.wire.WireIn; import org.apache.cassandra.cql3.QueryOptions; import org.apache.cassandra.transport.ProtocolVersion; +import static org.apache.cassandra.audit.FullQueryLogger.CURRENT_VERSION; import static org.apache.cassandra.audit.FullQueryLogger.GENERATED_NOW_IN_SECONDS; import static org.apache.cassandra.audit.FullQueryLogger.GENERATED_TIMESTAMP; import static org.apache.cassandra.audit.FullQueryLogger.KEYSPACE; import static org.apache.cassandra.audit.FullQueryLogger.PROTOCOL_VERSION; import static org.apache.cassandra.audit.FullQueryLogger.QUERY_OPTIONS; import static org.apache.cassandra.audit.FullQueryLogger.QUERY_START_TIME; -import static org.apache.cassandra.audit.FullQueryLogger.TYPE; -import static org.apache.cassandra.audit.FullQueryLogger.VERSION; import static org.apache.cassandra.audit.FullQueryLogger.BATCH; import static org.apache.cassandra.audit.FullQueryLogger.BATCH_TYPE; import static org.apache.cassandra.audit.FullQueryLogger.QUERIES; @@ -47,14 +46,18 @@ import static org.apache.cassandra.audit.FullQueryLogger.QUERY; import static org.apache.cassandra.audit.FullQueryLogger.SINGLE_QUERY; import static org.apache.cassandra.audit.FullQueryLogger.VALUES; +import static org.apache.cassandra.utils.binlog.BinLog.TYPE; +import static org.apache.cassandra.utils.binlog.BinLog.VERSION; + public class FQLQueryReader implements ReadMarshallable { private FQLQuery query; public void readMarshallable(WireIn wireIn) throws IORuntimeException { - int currentVersion = wireIn.read(VERSION).int16(); - String type = wireIn.read(TYPE).text(); + verifyVersion(wireIn); + String type = readType(wireIn); + long queryStartTime = wireIn.read(QUERY_START_TIME).int64(); int protocolVersion = wireIn.read(PROTOCOL_VERSION).int32(); QueryOptions queryOptions = QueryOptions.codec.decode(Unpooled.wrappedBuffer(wireIn.read(QUERY_OPTIONS).bytes()), ProtocolVersion.decode(protocolVersion, true)); @@ -105,12 +108,35 @@ public class FQLQueryReader implements ReadMarshallable values); break; default: - throw new RuntimeException("Unknown type: " + type); + throw new IORuntimeException("Unhandled record type: " + type); } } + private void verifyVersion(WireIn wireIn) + { + int version = wireIn.read(VERSION).int16(); + + if (version > CURRENT_VERSION) + { + throw new IORuntimeException("Unsupported record version [" + version + + "] - highest supported version is [" + CURRENT_VERSION + ']'); + } + } + + private String readType(WireIn wireIn) throws IORuntimeException + { + String type = wireIn.read(TYPE).text(); + if (!SINGLE_QUERY.equals(type) && !BATCH.equals(type)) + { + throw new IORuntimeException("Unsupported record type field [" + type + + "] - supported record types are [" + SINGLE_QUERY + ", " + BATCH + ']'); + } + + return type; + } + public FQLQuery getQuery() { return query; } -} \ No newline at end of file +} diff --git a/tools/fqltool/src/org/apache/cassandra/fqltool/commands/Dump.java b/tools/fqltool/src/org/apache/cassandra/fqltool/commands/Dump.java index b3e1f22747..46c731f943 100644 --- a/tools/fqltool/src/org/apache/cassandra/fqltool/commands/Dump.java +++ b/tools/fqltool/src/org/apache/cassandra/fqltool/commands/Dump.java @@ -31,6 +31,7 @@ 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.core.io.IORuntimeException; import net.openhft.chronicle.queue.ChronicleQueue; import net.openhft.chronicle.queue.ChronicleQueueBuilder; import net.openhft.chronicle.queue.ExcerptTailer; @@ -42,6 +43,7 @@ import net.openhft.chronicle.wire.WireIn; import org.apache.cassandra.audit.FullQueryLogger; import org.apache.cassandra.cql3.QueryOptions; import org.apache.cassandra.transport.ProtocolVersion; +import org.apache.cassandra.utils.binlog.BinLog; /** * Dump the contents of a list of paths containing full query logs @@ -73,11 +75,20 @@ public class Dump implements Runnable { sb.setLength(0); - int version = wireIn.read(FullQueryLogger.VERSION).int16(); - if (version != FullQueryLogger.CURRENT_VERSION) - throw new UnsupportedOperationException("Full query log of unexpected version " + version + " encountered"); + int version = wireIn.read(BinLog.VERSION).int16(); + if (version > FullQueryLogger.CURRENT_VERSION) + { + throw new IORuntimeException("Unsupported record version [" + version + + "] - highest supported version is [" + FullQueryLogger.CURRENT_VERSION + ']'); + } + + String type = wireIn.read(BinLog.TYPE).text(); + if (!FullQueryLogger.SINGLE_QUERY.equals((type)) && !FullQueryLogger.BATCH.equals((type))) + { + throw new IORuntimeException("Unsupported record type field [" + type + + "] - supported record types are [" + FullQueryLogger.SINGLE_QUERY + ", " + FullQueryLogger.BATCH + ']'); + } - String type = wireIn.read(FullQueryLogger.TYPE).text(); sb.append("Type: ") .append(type) .append(System.lineSeparator()); @@ -117,7 +128,7 @@ public class Dump implements Runnable break; default: - throw new UnsupportedOperationException("Log entry of unsupported type " + type); + throw new IORuntimeException("Log entry of unsupported type " + type); } System.out.print(sb.toString()); diff --git a/tools/fqltool/test/unit/org/apache/cassandra/fqltool/FQLReplayTest.java b/tools/fqltool/test/unit/org/apache/cassandra/fqltool/FQLReplayTest.java index 8c573f478a..9220c2ed07 100644 --- a/tools/fqltool/test/unit/org/apache/cassandra/fqltool/FQLReplayTest.java +++ b/tools/fqltool/test/unit/org/apache/cassandra/fqltool/FQLReplayTest.java @@ -36,10 +36,12 @@ import org.junit.Test; import com.datastax.driver.core.CodecRegistry; import com.datastax.driver.core.SimpleStatement; import com.datastax.driver.core.Statement; +import net.openhft.chronicle.core.io.IORuntimeException; import net.openhft.chronicle.queue.ChronicleQueue; import net.openhft.chronicle.queue.ChronicleQueueBuilder; import net.openhft.chronicle.queue.ExcerptAppender; import net.openhft.chronicle.queue.ExcerptTailer; +import net.openhft.chronicle.wire.WireOut; import org.apache.cassandra.audit.FullQueryLogger; import org.apache.cassandra.cql3.QueryOptions; import org.apache.cassandra.cql3.statements.BatchStatement; @@ -51,6 +53,7 @@ import org.apache.cassandra.tools.Util; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.MergeIterator; import org.apache.cassandra.utils.Pair; +import org.apache.cassandra.utils.binlog.BinLog; import static org.apache.cassandra.fqltool.QueryReplayer.ParsedTargetHost.fromString; import static org.junit.Assert.assertArrayEquals; @@ -586,6 +589,88 @@ public class FQLReplayTest fromString("aaa:bbb@abc.com:xyz"); } + @Test (expected = IORuntimeException.class) + public void testFutureVersion() throws Exception + { + FQLQueryReader reader = new FQLQueryReader(); + File dir = Files.createTempDirectory("chronicle").toFile(); + try (ChronicleQueue queue = ChronicleQueueBuilder.single(dir).build()) + { + ExcerptAppender appender = queue.acquireAppender(); + appender.writeDocument(new BinLog.ReleaseableWriteMarshallable() { + protected long version() + { + return 999; + } + + protected String type() + { + return FullQueryLogger.SINGLE_QUERY; + } + + public void writeMarshallablePayload(WireOut wire) + { + wire.write("future-field").text("future_value"); + } + + public void release() + { + + } + }); + + ExcerptTailer tailer = queue.createTailer(); + tailer.readDocument(reader); + } + catch (Exception e) + { + assertTrue(e.getMessage().contains("Unsupported record version")); + throw e; + } + + } + + @Test (expected = IORuntimeException.class) + public void testUnknownRecord() throws Exception + { + FQLQueryReader reader = new FQLQueryReader(); + File dir = Files.createTempDirectory("chronicle").toFile(); + try (ChronicleQueue queue = ChronicleQueueBuilder.single(dir).build()) + { + ExcerptAppender appender = queue.acquireAppender(); + appender.writeDocument(new BinLog.ReleaseableWriteMarshallable() { + protected long version() + { + return FullQueryLogger.CURRENT_VERSION; + } + + protected String type() + { + return "unknown-type"; + } + + public void writeMarshallablePayload(WireOut wire) + { + wire.write("unknown-field").text("unknown_value"); + } + + public void release() + { + + } + }); + + ExcerptTailer tailer = queue.createTailer(); + tailer.readDocument(reader); + } + catch (Exception e) + { + assertTrue(e.getMessage().contains("Unsupported record type field")); + throw e; + } + + } + private void compareStatements(Statement statement1, Statement statement2) { assertTrue(statement1 instanceof SimpleStatement && statement2 instanceof SimpleStatement);