From 99605d9f8db9a64ce58a738de5aa236103ec8f66 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Fri, 24 Aug 2012 14:57:37 -0500 Subject: [PATCH] add disk_failure_policy patch by Aleksey Yeschenko; reviewed by jbellis for CASSANDRA-2118 --- CHANGES.txt | 1 + NEWS.txt | 2 + conf/cassandra.yaml | 8 ++++ .../org/apache/cassandra/config/Config.java | 8 ++++ .../cassandra/config/DatabaseDescriptor.java | 7 ++++ .../cassandra/db/ColumnFamilyStore.java | 13 +++++- .../org/apache/cassandra/db/DataTracker.java | 26 ++++++++++++ .../org/apache/cassandra/db/Directories.java | 16 +++++--- src/java/org/apache/cassandra/db/Table.java | 17 ++++++++ .../db/columniterator/IndexedSliceReader.java | 2 +- .../columniterator/SSTableNamesIterator.java | 4 +- .../db/columniterator/SimpleSliceReader.java | 2 +- .../org/apache/cassandra/io/FSReadError.java | 5 ++- .../org/apache/cassandra/io/FSWriteError.java | 5 ++- .../io/sstable/CorruptSSTableException.java | 5 --- .../cassandra/io/sstable/Descriptor.java | 2 +- .../io/sstable/SSTableIdentityIterator.java | 1 - .../cassandra/io/sstable/SSTableReader.java | 2 - .../cassandra/service/CassandraDaemon.java | 40 ++++++++++++++++--- .../streaming/IncomingStreamReader.java | 2 +- 20 files changed, 136 insertions(+), 32 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index e3db234e13..2a10d98c8a 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -2,6 +2,7 @@ * Make compaction, flush JBOD-aware (CASSANDRA-4292) * run local range scans on the read stage (CASSANDRA-3687) * clean up ioexceptions (CASSANDRA-2116) + * add disk_failure_policy (CASSANDRA-2118) * Introduce new json format with row level deletion (CASSANDRA-4054) * remove redundant "name" column from schema_keyspaces (CASSANDRA-4433) * improve "nodetool ring" handling of multi-dc clusters (CASSANDRA-3047) diff --git a/NEWS.txt b/NEWS.txt index 199dd8ade2..7c64af212d 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -51,6 +51,8 @@ Features - num_tokens can now be specified in cassandra.yaml. This defines the number of tokens assigned to the host on the ring (default: 1). Also specifying initial_token will override any num_tokens setting. + - disk_failure_policy allows blacklisting failed disks in JBOD + configuration instead of erroring out indefinitely 1.1.3 diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 5e45961b97..b308b5c4e0 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -92,6 +92,14 @@ data_file_directories: # commit log commitlog_directory: /var/lib/cassandra/commitlog +# policy for data disk failures: +# stop: shut down gossip and Thrift, leaving the node effectively dead, but +# still inspectable via JMX. +# best_effort: stop using the failed disk and respond to requests based on +# remaining available sstables. This means you WILL see obsolete +# data at CL.ONE! +disk_failure_policy: stop + # Maximum size of the key cache in memory. # # Each key cache hit saves 1 seek and each row cache hit saves 2 seeks at the diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 8488d083a3..d66c79d4f7 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -40,6 +40,8 @@ public class Config public SeedProviderDef seed_provider; public DiskAccessMode disk_access_mode = DiskAccessMode.auto; + public DiskFailurePolicy disk_failure_policy = DiskFailurePolicy.best_effort; + /* initial token in the ring */ public String initial_token; public Integer num_tokens = 1; @@ -193,6 +195,12 @@ public class Config standard, } + public static enum DiskFailurePolicy + { + best_effort, + stop + } + public static enum RequestSchedulerId { keyspace diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index dce43c9486..ccd2b832bc 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -199,6 +199,8 @@ public class DatabaseDescriptor logger.info("DiskAccessMode is " + conf.disk_access_mode + ", indexAccessMode is " + indexAccessMode ); } + logger.info("DiskFailureMode is " + conf.disk_failure_policy); + logger.debug("page_cache_hinting is " + conf.populate_io_cache_on_flush); /* Authentication and authorization backend, implementing IAuthenticator and IAuthority */ @@ -996,6 +998,11 @@ public class DatabaseDescriptor return indexAccessMode; } + public static Config.DiskFailurePolicy getDiskFailureMode() + { + return conf.disk_failure_policy; + } + public static boolean isSnapshotBeforeCompaction() { return conf.snapshot_before_compaction; diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index c3c5f8e2df..f31276f236 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -229,7 +229,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean if (loadSSTables) { Directories.SSTableLister sstableFiles = directories.sstableLister().skipTemporary(true); - Collection sstables = SSTableReader.batchOpen(sstableFiles.list().entrySet(), data, metadata, this.partitioner); + Collection sstables = SSTableReader.batchOpen(sstableFiles.list().entrySet(), metadata, this.partitioner); // Filter non-compacted sstables, remove compacted ones Set compactedSSTables = new HashSet(); @@ -293,6 +293,15 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean } } + /** + * Removes every SSTable in the directory from the DataTracker's view. + * @param directory the unreadable directory, possibly with SSTables in it, but not necessarily. + */ + void maybeRemoveUnreadableSSTables(File directory) + { + data.removeUnreadableSSTables(directory); + } + void unregisterMBean() throws MalformedObjectNameException, InstanceNotFoundException, MBeanRegistrationException { MBeanServer mbs = ManagementFactory.getPlatformMBeanServer(); @@ -1255,7 +1264,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean // if (view.intervalTree.isEmpty()) { - sstables = Collections.emptyList(); + sstables = Collections.emptyList(); break; } diff --git a/src/java/org/apache/cassandra/db/DataTracker.java b/src/java/org/apache/cassandra/db/DataTracker.java index 1061a798ab..8233fbe479 100644 --- a/src/java/org/apache/cassandra/db/DataTracker.java +++ b/src/java/org/apache/cassandra/db/DataTracker.java @@ -290,6 +290,32 @@ public class DataTracker postReplace(notCompacting, Collections.emptySet()); } + /** + * Removes every SSTable in the directory from the DataTracker's view. + * @param directory the unreadable directory, possibly with SSTables in it, but not necessarily. + */ + void removeUnreadableSSTables(File directory) + { + View currentView, newView; + List remaining = new ArrayList(); + do + { + currentView = view.get(); + for (SSTableReader r : currentView.nonCompactingSStables()) + { + if (!r.descriptor.directory.equals(directory)) + remaining.add(r); + } + + if (remaining.size() == currentView.nonCompactingSStables().size()) + return; + + newView = currentView.replace(currentView.sstables, remaining); + } + while (!view.compareAndSet(currentView, newView)); + notifySSTablesChanged(remaining, Collections.emptySet(), OperationType.UNKNOWN); + } + /** (Re)initializes the tracker, purging all references. */ void init() { diff --git a/src/java/org/apache/cassandra/db/Directories.java b/src/java/org/apache/cassandra/db/Directories.java index a6fac8f511..ee53fd2f16 100644 --- a/src/java/org/apache/cassandra/db/Directories.java +++ b/src/java/org/apache/cassandra/db/Directories.java @@ -162,9 +162,13 @@ public class Directories for (File dir : sstableDirectories) { - if (maxFreeDisk < dir.getUsableSpace()) + if (BlacklistedDirectories.isUnwritable(dir)) + continue; + + long usableSpace = dir.getUsableSpace(); + if (maxFreeDisk < usableSpace) { - maxFreeDisk = dir.getUsableSpace(); + maxFreeDisk = usableSpace; maxLocation = dir; } } @@ -173,10 +177,7 @@ public class Directories logger.debug(String.format("expected data files size is %d; largest free partition (%s) has %d bytes free", estimatedSize, maxLocation, maxFreeDisk)); - - if (estimatedSize < maxFreeDisk) - return maxLocation; - return null; + return estimatedSize < maxFreeDisk ? maxLocation : null; } /** @@ -323,6 +324,9 @@ public class Directories for (File location : sstableDirectories) { + if (BlacklistedDirectories.isUnreadable(location)) + continue; + if (snapshotName != null) { new File(location, join(SNAPSHOT_SUBDIR, snapshotName)).listFiles(getFilter()); diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java index 0732534110..e41b11f65b 100644 --- a/src/java/org/apache/cassandra/db/Table.java +++ b/src/java/org/apache/cassandra/db/Table.java @@ -17,6 +17,7 @@ */ package org.apache.cassandra.db; +import java.io.File; import java.io.IOException; import java.nio.ByteBuffer; import java.util.*; @@ -125,6 +126,22 @@ public class Table } } + /** + * Removes every SSTable in the directory from the appropriate DataTracker's view. + * @param directory the unreadable directory, possibly with SSTables in it, but not necessarily. + */ + public static void removeUnreadableSSTables(File directory) + { + for (Table table : Table.all()) + { + for (ColumnFamilyStore baseCfs : table.getColumnFamilyStores()) + { + for (ColumnFamilyStore cfs : baseCfs.concatWithIndexes()) + cfs.maybeRemoveUnreadableSSTables(directory); + } + } + } + public Collection getColumnFamilyStores() { return Collections.unmodifiableCollection(columnFamilyStores.values()); diff --git a/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java b/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java index 98a95527da..db0130e519 100644 --- a/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java +++ b/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java @@ -107,7 +107,7 @@ class IndexedSliceReader extends AbstractIterator implements OnDiskA catch (IOException e) { sstable.markSuspect(); - throw new CorruptSSTableException(e, sstable.descriptor); + throw new CorruptSSTableException(e, file.getPath()); } } diff --git a/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java b/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java index 05d2bc0ad3..2cb3435664 100644 --- a/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java +++ b/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java @@ -66,7 +66,7 @@ public class SSTableNamesIterator extends SimpleAbstractColumnIterator implement catch (IOException e) { sstable.markSuspect(); - throw new CorruptSSTableException(e, sstable.descriptor); + throw new CorruptSSTableException(e, sstable.getFilename()); } finally { @@ -89,7 +89,7 @@ public class SSTableNamesIterator extends SimpleAbstractColumnIterator implement catch (IOException e) { sstable.markSuspect(); - throw new CorruptSSTableException(e, sstable.descriptor); + throw new CorruptSSTableException(e, sstable.getFilename()); } } diff --git a/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java b/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java index b9a5a64fea..d19e6a503f 100644 --- a/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java +++ b/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java @@ -86,7 +86,7 @@ class SimpleSliceReader extends AbstractIterator implements OnDiskAt catch (IOException e) { sstable.markSuspect(); - throw new CorruptSSTableException(e, sstable.descriptor); + throw new CorruptSSTableException(e, sstable.getFilename()); } } diff --git a/src/java/org/apache/cassandra/io/FSReadError.java b/src/java/org/apache/cassandra/io/FSReadError.java index c180eb06b1..1f208c0566 100644 --- a/src/java/org/apache/cassandra/io/FSReadError.java +++ b/src/java/org/apache/cassandra/io/FSReadError.java @@ -33,8 +33,9 @@ public class FSReadError extends FSError this(cause, new File(path)); } - public FSReadError(Throwable cause, Descriptor descriptor) + @Override + public String toString() { - this(cause, descriptor.baseFilename()); + return "FSReadError in " + path; } } diff --git a/src/java/org/apache/cassandra/io/FSWriteError.java b/src/java/org/apache/cassandra/io/FSWriteError.java index 9346f62aec..4b68f1b0eb 100644 --- a/src/java/org/apache/cassandra/io/FSWriteError.java +++ b/src/java/org/apache/cassandra/io/FSWriteError.java @@ -33,8 +33,9 @@ public class FSWriteError extends FSError this(cause, new File(path)); } - public FSWriteError(Throwable cause, Descriptor descriptor) + @Override + public String toString() { - this(cause, descriptor.baseFilename()); + return "FSWriteError in " + path; } } diff --git a/src/java/org/apache/cassandra/io/sstable/CorruptSSTableException.java b/src/java/org/apache/cassandra/io/sstable/CorruptSSTableException.java index 61cead4222..a71daaf44d 100644 --- a/src/java/org/apache/cassandra/io/sstable/CorruptSSTableException.java +++ b/src/java/org/apache/cassandra/io/sstable/CorruptSSTableException.java @@ -33,9 +33,4 @@ public class CorruptSSTableException extends RuntimeException { this(cause, new File(path)); } - - public CorruptSSTableException(Exception cause, Descriptor descriptor) - { - this(cause, descriptor.baseFilename()); - } } diff --git a/src/java/org/apache/cassandra/io/sstable/Descriptor.java b/src/java/org/apache/cassandra/io/sstable/Descriptor.java index d03c2e8fc9..4f776d616c 100644 --- a/src/java/org/apache/cassandra/io/sstable/Descriptor.java +++ b/src/java/org/apache/cassandra/io/sstable/Descriptor.java @@ -204,7 +204,7 @@ public class Descriptor return filenameFor(component.name()); } - public String baseFilename() + private String baseFilename() { StringBuilder buff = new StringBuilder(); buff.append(directory).append(File.separatorChar); diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java index 337d6b61a2..f5a9421dd2 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java @@ -77,7 +77,6 @@ public class SSTableIdentityIterator implements Comparable batchOpen(Set>> entries, - final DataTracker tracker, final CFMetaData metadata, final IPartitioner partitioner) { diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index 11e8a24e8c..5a6042b075 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -34,11 +34,10 @@ import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.ConfigurationException; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.Schema; -import org.apache.cassandra.db.ColumnFamilyStore; -import org.apache.cassandra.db.Directories; -import org.apache.cassandra.db.SystemTable; -import org.apache.cassandra.db.Table; +import org.apache.cassandra.db.*; import org.apache.cassandra.db.commitlog.CommitLog; +import org.apache.cassandra.io.FSError; +import org.apache.cassandra.io.FSReadError; import org.apache.cassandra.thrift.ThriftServer; import org.apache.cassandra.utils.CLibrary; import org.apache.cassandra.utils.Mx4jTool; @@ -134,6 +133,37 @@ public class CassandraDaemon // some code, like FileChannel.map, will wrap an OutOfMemoryError in another exception if (e2 instanceof OutOfMemoryError) System.exit(100); + + if (e2 instanceof FSError) + { + if (e2 != e) // make sure FSError gets logged exactly once. + logger.error("Exception in thread " + t, e2); + handleFSError((FSError) e2); + } + } + } + + private void handleFSError(FSError e) + { + switch (DatabaseDescriptor.getDiskFailureMode()) + { + case stop: + logger.error("Stopping the gossiper and the RPC server"); + StorageService.instance.stopGossiping(); + StorageService.instance.stopRPCServer(); + break; + case best_effort: + // for both read and write errors mark the path as unwritable. + BlacklistedDirectories.maybeMarkUnwritable(e.path); + if (e instanceof FSReadError) + { + File directory = BlacklistedDirectories.maybeMarkUnreadable(e.path); + if (directory != null) + Table.removeUnreadableSSTables(directory); + } + break; + default: + throw new IllegalStateException(); } } }); @@ -266,8 +296,6 @@ public class CassandraDaemon * initialized via {@link #init(String[])} * * Hook for JSVC - * - * @throws IOException */ public void start() { diff --git a/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java b/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java index 1e5553aaac..f74f566cbf 100644 --- a/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java +++ b/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java @@ -145,7 +145,7 @@ public class IncomingStreamReader { // need to update row cache // Note: Because we won't just echo the columns, there is no need to use the PRESERVE_SIZE flag, contrarily to what appendFromStream does below - SSTableIdentityIterator iter = new SSTableIdentityIterator(cfs.metadata, in, localFile.desc.baseFilename(), key, 0, dataSize, IColumnSerializer.Flag.FROM_REMOTE); + SSTableIdentityIterator iter = new SSTableIdentityIterator(cfs.metadata, in, localFile.getFilename(), key, 0, dataSize, IColumnSerializer.Flag.FROM_REMOTE); PrecompactedRow row = new PrecompactedRow(controller, Collections.singletonList(iter)); // We don't expire anything so the row shouldn't be empty assert !row.isEmpty();