add disk_failure_policy

patch by Aleksey Yeschenko; reviewed by jbellis for CASSANDRA-2118

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)

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

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

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

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;

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<SSTableReader> sstables = SSTableReader.batchOpen(sstableFiles.list().entrySet(), data, metadata, this.partitioner);
+            Collection<SSTableReader> sstables = SSTableReader.batchOpen(sstableFiles.list().entrySet(), metadata, this.partitioner);
 
             // Filter non-compacted sstables, remove compacted ones
             Set<Integer> compactedSSTables = new HashSet<Integer>();
@@ -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.<SSTableReader>emptyList();
+                sstables = Collections.emptyList();
                 break;
             }
 

src/java/org/apache/cassandra/db/DataTracker.java

@@ -290,6 +290,32 @@ public class DataTracker
         postReplace(notCompacting, Collections.<SSTableReader>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<SSTableReader> remaining = new ArrayList<SSTableReader>();
+        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.<SSTableReader>emptySet(), OperationType.UNKNOWN);
+    }
+
     /** (Re)initializes the tracker, purging all references. */
     void init()
     {

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());

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<ColumnFamilyStore> getColumnFamilyStores()
     {
         return Collections.unmodifiableCollection(columnFamilyStores.values());

src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java

@@ -107,7 +107,7 @@ class IndexedSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskA
         catch (IOException e)
         {
             sstable.markSuspect();
-            throw new CorruptSSTableException(e, sstable.descriptor);
+            throw new CorruptSSTableException(e, file.getPath());
         }
     }
 

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());
         }
     }
 

src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java

@@ -86,7 +86,7 @@ class SimpleSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskAt
         catch (IOException e)
         {
             sstable.markSuspect();
-            throw new CorruptSSTableException(e, sstable.descriptor);
+            throw new CorruptSSTableException(e, sstable.getFilename());
         }
     }
 

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;
     }
 }

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;
     }
 }

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());
-    }
 }

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

src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java

@@ -77,7 +77,6 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
      * @param dataStart Data for this row starts at this pos.
      * @param dataSize length of row data
      * @param checkData if true, do its best to deserialize and check the coherence of row data
-     * @throws IOException
      */
     public SSTableIdentityIterator(SSTableReader sstable, RandomAccessReader file, DecoratedKey key, long dataStart, long dataSize, boolean checkData)
     {

src/java/org/apache/cassandra/io/sstable/SSTableReader.java

@@ -38,7 +38,6 @@ import org.apache.cassandra.db.*;
 import org.apache.cassandra.db.commitlog.ReplayPosition;
 import org.apache.cassandra.db.index.SecondaryIndex;
 import org.apache.cassandra.db.filter.QueryFilter;
-import org.apache.cassandra.db.index.keys.KeysIndex;
 import org.apache.cassandra.dht.AbstractBounds;
 import org.apache.cassandra.dht.IPartitioner;
 import org.apache.cassandra.dht.LocalPartitioner;
@@ -217,7 +216,6 @@ public class SSTableReader extends SSTable
     }
 
     public static Collection<SSTableReader> batchOpen(Set<Map.Entry<Descriptor, Set<Component>>> entries,
-                                                      final DataTracker tracker,
                                                       final CFMetaData metadata,
                                                       final IPartitioner partitioner)
     {

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()
     {

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();