natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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
This commit is contained in:
Stefan Miklosovic 2022-10-11 23:13:13 +02:00
parent e13356d75d
commit 488c0c75a8
3 changed files with 207 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

81
src/java/org/apache/cassandra/utils/binlog/BinLog.java
View File

@ -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<File, File[]> 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;
}
}

145
test/unit/org/apache/cassandra/audit/AuditLoggerCleanupTest.java Normal file
View File

@ -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<Path> 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);
}
}
}
}