From 488c0c75a8f632f2db4e3db39f2ebcf8a489971e Mon Sep 17 00:00:00 2001 From: Stefan Miklosovic Date: Tue, 11 Oct 2022 23:13:13 +0200 Subject: [PATCH] Remove empty cq4 log files to prevent BinLog from failing to start This patch also backports CASSANDRA-17595. patch by Stefan Miklosovic; reviewed by Caleb Rackliffe for CASSANDRA-17933 --- CHANGES.txt | 1 + .../apache/cassandra/utils/binlog/BinLog.java | 81 +++++++--- .../audit/AuditLoggerCleanupTest.java | 145 ++++++++++++++++++ 3 files changed, 207 insertions(+), 20 deletions(-) create mode 100644 test/unit/org/apache/cassandra/audit/AuditLoggerCleanupTest.java diff --git a/CHANGES.txt b/CHANGES.txt index 1907e6cae7..4a85c97c53 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0.7 + * Remove empty cq4 files in log directory to not fail the startup of BinLog (CASSANDRA-17933) * Fix multiple BufferPool bugs (CASSANDRA-16681) * Fix StorageService.getNativeaddress handling of IPv6 addresses (CASSANDRA-17945) * Mitigate direct buffer memory OOM on replacements (CASSANDRA-17895) diff --git a/src/java/org/apache/cassandra/utils/binlog/BinLog.java b/src/java/org/apache/cassandra/utils/binlog/BinLog.java index 63f74a35d8..b2267512ac 100644 --- a/src/java/org/apache/cassandra/utils/binlog/BinLog.java +++ b/src/java/org/apache/cassandra/utils/binlog/BinLog.java @@ -20,7 +20,6 @@ package org.apache.cassandra.utils.binlog; import java.io.File; import java.nio.file.Path; -import java.nio.file.Paths; import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; @@ -28,6 +27,7 @@ import java.util.List; import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; @@ -36,13 +36,13 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import net.openhft.chronicle.queue.ChronicleQueue; +import net.openhft.chronicle.queue.impl.single.SingleChronicleQueue; import net.openhft.chronicle.queue.impl.single.SingleChronicleQueueBuilder; import net.openhft.chronicle.queue.ExcerptAppender; import net.openhft.chronicle.queue.RollCycles; import net.openhft.chronicle.wire.WireOut; import net.openhft.chronicle.wire.WriteMarshallable; import org.apache.cassandra.concurrent.NamedThreadFactory; -import org.apache.cassandra.fql.FullQueryLoggerOptions; import org.apache.cassandra.io.FSError; import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.utils.JVMStabilityInspector; @@ -50,6 +50,8 @@ import org.apache.cassandra.utils.NoSpamLogger; import org.apache.cassandra.utils.Throwables; import org.apache.cassandra.utils.concurrent.WeightedQueue; +import static java.lang.String.format; + /** * 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 @@ -426,6 +428,12 @@ public class BinLog implements Runnable } try { + Throwable sanitationThrowable = cleanEmptyLogFiles(path.toFile(), null); + if (sanitationThrowable != null) + throw new RuntimeException(format("Unable to clean up %s directory from empty %s files.", + path.toAbsolutePath(), SingleChronicleQueue.SUFFIX), + sanitationThrowable); + // create the archiver before cleaning directories - ExternalArchiver will try to archive any existing file. BinLogArchiver archiver = Strings.isNullOrEmpty(archiveCommand) ? new DeletingArchiver(maxLogSize) : new ExternalArchiver(archiveCommand, path, maxArchiveRetries); if (cleanDirectory) @@ -462,24 +470,46 @@ public class BinLog implements Runnable } } + /** + * ChronicleQueue fails to start on cq4 files which are empty. Find such files in log dir and remove them. + */ + private static Throwable cleanEmptyLogFiles(File directory, Throwable accumulate) + { + return cleanDirectory(directory, accumulate, + (dir) -> dir.listFiles(file -> { + boolean foundEmptyCq4File = !file.isDirectory() + && file.length() == 0 + && file.getName().endsWith(SingleChronicleQueue.SUFFIX); + + if (foundEmptyCq4File) + logger.warn("Found empty ChronicleQueue file {}. This file wil be deleted as part of BinLog initialization.", + file.getAbsolutePath()); + + return foundEmptyCq4File; + })); + } + public static Throwable cleanDirectory(File directory, Throwable accumulate) { - if (!directory.exists()) - { - return Throwables.merge(accumulate, new RuntimeException(String.format("%s does not exists", directory))); - } - 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); - } + return cleanDirectory(directory, accumulate, File::listFiles); + } + + private static Throwable cleanDirectory(File directory, Throwable accumulate, Function lister) + { + accumulate = checkDirectory(directory, accumulate); + + if (accumulate != null) + return accumulate; + + File[] files = lister.apply(directory); + + if (files != null) + for (File f : files) + accumulate = deleteRecursively(f, accumulate); + if (accumulate instanceof FSError) - { JVMStabilityInspector.inspectThrowable(accumulate); - } + return accumulate; } @@ -487,11 +517,22 @@ public class BinLog implements Runnable { if (fileOrDirectory.isDirectory()) { - for (File f : fileOrDirectory.listFiles()) - { - accumulate = FileUtils.deleteWithConfirm(f, accumulate); - } + File[] files = fileOrDirectory.listFiles(); + if (files != null) + for (File f : files) + accumulate = FileUtils.deleteWithConfirm(f, accumulate); } return FileUtils.deleteWithConfirm(fileOrDirectory, accumulate); } + + private static Throwable checkDirectory(File directory, Throwable accumulate) + { + if (!directory.exists()) + accumulate = Throwables.merge(accumulate, new RuntimeException(format("%s does not exist", directory))); + + if (!directory.isDirectory()) + accumulate = Throwables.merge(accumulate, new RuntimeException(format("%s is not a directory", directory))); + + return accumulate; + } } diff --git a/test/unit/org/apache/cassandra/audit/AuditLoggerCleanupTest.java b/test/unit/org/apache/cassandra/audit/AuditLoggerCleanupTest.java new file mode 100644 index 0000000000..891827cd3e --- /dev/null +++ b/test/unit/org/apache/cassandra/audit/AuditLoggerCleanupTest.java @@ -0,0 +1,145 @@ +/* + * 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.audit; + +import java.io.File; +import java.net.InetAddress; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.UUID; +import java.util.stream.Stream; + +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import com.datastax.driver.core.Cluster; +import com.datastax.driver.core.Session; +import net.openhft.chronicle.queue.impl.single.SingleChronicleQueue; +import net.openhft.chronicle.queue.impl.table.SingleTableStore; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.config.OverrideConfigurationLoader; +import org.apache.cassandra.service.EmbeddedCassandraService; +import org.apache.cassandra.service.StorageService; + +import static java.nio.file.Files.list; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class AuditLoggerCleanupTest +{ + @Rule + public TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private EmbeddedCassandraService embedded; + private File auditLogDirRoot; + private File emptyCq4File; + private File emptyMetadataFile; + + @Before + public void setup() throws Exception + { + OverrideConfigurationLoader.override((config) -> { + config.audit_logging_options.enabled = true; + config.audit_logging_options.audit_logs_dir = auditLogDirRoot.getAbsolutePath(); + config.audit_logging_options.roll_cycle = "MINUTELY"; + }); + + auditLogDirRoot = temporaryFolder.getRoot(); + + // invalid file will be removed on startup + emptyCq4File = Files.createFile(auditLogDirRoot.toPath().resolve("20220928-12" + SingleChronicleQueue.SUFFIX)).toFile(); + emptyMetadataFile = Files.createFile(auditLogDirRoot.toPath().resolve("metadata" + SingleTableStore.SUFFIX)).toFile(); + + System.setProperty("cassandra.superuser_setup_delay_ms", "0"); + embedded = new EmbeddedCassandraService(); + embedded.start(); + } + + @After + public void shutdown() + { + embedded.stop(); + } + + @Test + public void testCleanupOfAuditLogDir() throws Throwable + { + // node started even there was empty cq4 file as it was removed upon start + assertTrue(StorageService.instance.isAuditLogEnabled()); + assertFalse(emptyCq4File.exists()); + // empty metadata file is reused + assertTrue(emptyMetadataFile.exists() && emptyMetadataFile.length() != 0); + + insertData(); + + assertLogFileExists(); + + StorageService.instance.disableAuditLog(); + + // disabling and enabling from JMX will trigger same empty file cleanup + + emptyCq4File = Files.createFile(auditLogDirRoot.toPath().resolve("20220928-12" + SingleChronicleQueue.SUFFIX)).toFile(); + + StorageService.instance.enableAuditLog(null, null, null, null, null, null, null, null); + + assertTrue(StorageService.instance.isAuditLogEnabled()); + + // invalid file were removed again + assertFalse(emptyCq4File.exists()); + // only valid files are present + assertLogFileExists(); + } + + private void assertLogFileExists() throws Exception + { + try (Stream stream = list(auditLogDirRoot.toPath())) + { + assertTrue(stream.allMatch(p -> { + String fileName = p.getFileName().toString(); + return (fileName.endsWith(SingleChronicleQueue.SUFFIX) || fileName.endsWith(SingleTableStore.SUFFIX)) + && p.toFile().isFile() && p.toFile().length() != 0; + })); + } + } + + private void insertData() + { + String keyspaceName = "c17933_" + UUID.randomUUID().toString().replace("-", ""); + execute("CREATE KEYSPACE " + keyspaceName + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'};"); + execute("CREATE TABLE " + keyspaceName + ".tb1 (id int primary key);"); + execute("INSERT INTO " + keyspaceName + ".tb1 (id) VALUES (1)"); + } + + private static void execute(String query) + { + try ( + Cluster cluster = Cluster.builder().addContactPoints(InetAddress.getLoopbackAddress()) + .withoutJMXReporting() + .withPort(DatabaseDescriptor.getNativeTransportPort()).build()) + { + try (Session session = cluster.connect()) + { + session.execute(query); + } + } + } +}