hadoop-mapreduce-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sc...@apache.org
Subject svn commit: r1026137 - in /hadoop/mapreduce/trunk: ./ src/contrib/raid/src/java/org/apache/hadoop/hdfs/ src/contrib/raid/src/java/org/apache/hadoop/hdfs/server/ src/contrib/raid/src/java/org/apache/hadoop/hdfs/server/datanode/ src/contrib/raid/src/java...
Date Thu, 21 Oct 2010 20:48:46 GMT
Author: schen
Date: Thu Oct 21 20:48:45 2010
New Revision: 1026137

URL: http://svn.apache.org/viewvc?rev=1026137&view=rev
Log:
MAPREDUCE-2132. A command line option in RaidShell to fix blocks using raid

Added:
    hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/hdfs/RaidDFSUtil.java
    hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/hdfs/server/
    hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/hdfs/server/datanode/
    hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/hdfs/server/datanode/RaidBlockSender.java
    hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/BlockFixer.java
    hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/HarIndex.java
    hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/raid/TestHarIndexParser.java
    hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidShell.java
Modified:
    hadoop/mapreduce/trunk/CHANGES.txt
    hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidNode.java
    hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidShell.java
    hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidUtils.java
    hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/hdfs/TestRaidDfs.java
    hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidNode.java

Modified: hadoop/mapreduce/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/CHANGES.txt?rev=1026137&r1=1026136&r2=1026137&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/CHANGES.txt (original)
+++ hadoop/mapreduce/trunk/CHANGES.txt Thu Oct 21 20:48:45 2010
@@ -138,6 +138,8 @@ Trunk (unreleased changes)
     MAPREDUCE-1819. RaidNode is now smarter in submitting Raid jobs. (Ramkumar
     Vadali via schen)
 
+    MAPREDUCE-2132. A command line option in RaidShell to fix blocks using raid
+
   OPTIMIZATIONS
 
     MAPREDUCE-1354. Enhancements to JobTracker for better performance and

Added: hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/hdfs/RaidDFSUtil.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/hdfs/RaidDFSUtil.java?rev=1026137&view=auto
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/hdfs/RaidDFSUtil.java (added)
+++ hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/hdfs/RaidDFSUtil.java Thu Oct 21 20:48:45 2010
@@ -0,0 +1,52 @@
+/**
+ * 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.hadoop.hdfs;
+
+import java.io.IOException;
+import java.util.LinkedList;
+import java.util.List;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.hdfs.protocol.LocatedBlock;
+import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
+
+public abstract class RaidDFSUtil {
+  /**
+   * Returns the corrupt blocks in a file.
+   */
+  public static List<LocatedBlock> corruptBlocksInFile(
+    DistributedFileSystem dfs, String path, long offset, long length)
+  throws IOException {
+    List<LocatedBlock> corrupt = new LinkedList<LocatedBlock>();
+    LocatedBlocks locatedBlocks =
+      getBlockLocations(dfs, path, offset, length);
+    for (LocatedBlock b: locatedBlocks.getLocatedBlocks()) {
+      if (b.isCorrupt() ||
+         (b.getLocations().length == 0 && b.getBlockSize() > 0)) {
+        corrupt.add(b);
+      }
+    }
+    return corrupt;
+  }
+
+  public static LocatedBlocks getBlockLocations(
+    DistributedFileSystem dfs, String path, long offset, long length)
+    throws IOException {
+    return dfs.getClient().namenode.getBlockLocations(path, offset, length);
+  }
+}

Added: hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/hdfs/server/datanode/RaidBlockSender.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/hdfs/server/datanode/RaidBlockSender.java?rev=1026137&view=auto
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/hdfs/server/datanode/RaidBlockSender.java (added)
+++ hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/hdfs/server/datanode/RaidBlockSender.java Thu Oct 21 20:48:45 2010
@@ -0,0 +1,477 @@
+/**
+ * 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.hadoop.hdfs.server.datanode;
+
+import java.io.BufferedInputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.SocketException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.util.Arrays;
+
+import org.apache.commons.logging.Log;
+import org.apache.hadoop.fs.ChecksumException;
+import org.apache.hadoop.hdfs.protocol.Block;
+import org.apache.hadoop.hdfs.protocol.FSConstants;
+import org.apache.hadoop.hdfs.protocol.DataTransferProtocol.PacketHeader;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.net.SocketOutputStream;
+import org.apache.hadoop.util.DataChecksum;
+import org.apache.hadoop.util.StringUtils;
+
+/**
+ * Reads a block from the disk and sends it to a recipient.
+ */
+public class RaidBlockSender implements java.io.Closeable, FSConstants {
+  public static final Log LOG = DataNode.LOG;
+  static final Log ClientTraceLog = DataNode.ClientTraceLog;
+  
+  private Block block; // the block to read from
+
+  /** the replica to read from */
+  private final Replica replica = null;
+  /** The visible length of a replica. */
+  private final long replicaVisibleLength;
+
+  private InputStream blockIn; // data stream
+  private long blockInPosition = -1; // updated while using transferTo().
+  private DataInputStream checksumIn; // checksum datastream
+  private DataChecksum checksum; // checksum stream
+  private long offset; // starting position to read
+  private long endOffset; // ending position
+  private int bytesPerChecksum; // chunk size
+  private int checksumSize; // checksum size
+  private boolean corruptChecksumOk; // if need to verify checksum
+  private boolean chunkOffsetOK; // if need to send chunk offset
+  private long seqno; // sequence number of packet
+
+  private boolean transferToAllowed = true;
+  private boolean blockReadFully; //set when the whole block is read
+  private boolean verifyChecksum; //if true, check is verified while reading
+  private BlockTransferThrottler throttler;
+  private final String clientTraceFmt; // format of client trace log message
+
+  /**
+   * Minimum buffer used while sending data to clients. Used only if
+   * transferTo() is enabled. 64KB is not that large. It could be larger, but
+   * not sure if there will be much more improvement.
+   */
+  private static final int MIN_BUFFER_WITH_TRANSFERTO = 64*1024;
+  private volatile ChunkChecksum lastChunkChecksum = null;
+
+  
+  public RaidBlockSender(Block block, long blockLength, long startOffset, long length,
+              boolean corruptChecksumOk, boolean chunkOffsetOK,
+              boolean verifyChecksum, boolean transferToAllowed,
+              DataInputStream metadataIn, InputStreamFactory streamFactory
+              ) throws IOException {
+    this(block, blockLength, startOffset, length,
+        corruptChecksumOk, chunkOffsetOK,
+        verifyChecksum, transferToAllowed,
+        metadataIn, streamFactory, null);
+  }
+
+  public RaidBlockSender(Block block, long blockLength, long startOffset, long length,
+              boolean corruptChecksumOk, boolean chunkOffsetOK,
+              boolean verifyChecksum, boolean transferToAllowed,
+              DataInputStream metadataIn, InputStreamFactory streamFactory,
+              String clientTraceFmt) throws IOException {
+    try {
+      this.block = block;
+      ChunkChecksum chunkChecksum = null;
+      this.chunkOffsetOK = chunkOffsetOK;
+      this.corruptChecksumOk = corruptChecksumOk;
+      this.verifyChecksum = verifyChecksum;
+      this.replicaVisibleLength = blockLength;
+      this.transferToAllowed = transferToAllowed;
+      this.clientTraceFmt = clientTraceFmt;
+
+      if ( !corruptChecksumOk || metadataIn != null) {
+        this.checksumIn = metadataIn;
+
+        // read and handle the common header here. For now just a version
+       BlockMetadataHeader header = BlockMetadataHeader.readHeader(checksumIn);
+       short version = header.getVersion();
+
+        if (version != FSDataset.METADATA_VERSION) {
+          LOG.warn("Wrong version (" + version + ") for metadata file for "
+              + block + " ignoring ...");
+        }
+        checksum = header.getChecksum();
+      } else {
+        LOG.warn("Could not find metadata file for " + block);
+        // This only decides the buffer size. Use BUFFER_SIZE?
+        checksum = DataChecksum.newDataChecksum(DataChecksum.CHECKSUM_NULL,
+            16 * 1024);
+      }
+
+      /* If bytesPerChecksum is very large, then the metadata file
+       * is mostly corrupted. For now just truncate bytesPerchecksum to
+       * blockLength.
+       */        
+      bytesPerChecksum = checksum.getBytesPerChecksum();
+      if (bytesPerChecksum > 10*1024*1024 && bytesPerChecksum > replicaVisibleLength) {
+        checksum = DataChecksum.newDataChecksum(checksum.getChecksumType(),
+            Math.max((int)replicaVisibleLength, 10*1024*1024));
+        bytesPerChecksum = checksum.getBytesPerChecksum();        
+      }
+      checksumSize = checksum.getChecksumSize();
+
+      if (length < 0) {
+        length = replicaVisibleLength;
+      }
+
+      // end is either last byte on disk or the length for which we have a 
+      // checksum
+      if (chunkChecksum != null) {
+        endOffset = chunkChecksum.getDataLength();
+      } else {
+        endOffset = blockLength;
+      }
+      
+      if (startOffset < 0 || startOffset > endOffset
+          || (length + startOffset) > endOffset) {
+        String msg = " Offset " + startOffset + " and length " + length
+        + " don't match block " + block + " ( blockLen " + endOffset + " )";
+        LOG.warn("sendBlock() : " + msg);
+        throw new IOException(msg);
+      }
+      
+      offset = (startOffset - (startOffset % bytesPerChecksum));
+      if (length >= 0) {
+        // Make sure endOffset points to end of a checksumed chunk.
+        long tmpLen = startOffset + length;
+        if (tmpLen % bytesPerChecksum != 0) {
+          tmpLen += (bytesPerChecksum - tmpLen % bytesPerChecksum);
+        }
+        if (tmpLen < endOffset) {
+          // will use on-disk checksum here since the end is a stable chunk
+          endOffset = tmpLen;
+        } else if (chunkChecksum != null) {
+          //in last chunk which is changing. flag that we need to use in-memory 
+          // checksum 
+          this.lastChunkChecksum = chunkChecksum;
+        }
+      }
+
+      // seek to the right offsets
+      if (offset > 0) {
+        long checksumSkip = (offset / bytesPerChecksum) * checksumSize;
+        // note blockInStream is seeked when created below
+        if (checksumSkip > 0) {
+          // Should we use seek() for checksum file as well?
+          IOUtils.skipFully(checksumIn, checksumSkip);
+        }
+      }
+      seqno = 0;
+      
+      blockIn = streamFactory.createStream(offset);
+    } catch (IOException ioe) {
+      IOUtils.closeStream(this);
+      IOUtils.closeStream(blockIn);
+      throw ioe;
+    }
+  }
+
+  /**
+   * close opened files.
+   */
+  public void close() throws IOException {
+    IOException ioe = null;
+    // close checksum file
+    if(checksumIn!=null) {
+      try {
+        checksumIn.close();
+      } catch (IOException e) {
+        ioe = e;
+      }
+      checksumIn = null;
+    }
+    // close data file
+    if(blockIn!=null) {
+      try {
+        blockIn.close();
+      } catch (IOException e) {
+        ioe = e;
+      }
+      blockIn = null;
+    }
+    // throw IOException if there is any
+    if(ioe!= null) {
+      throw ioe;
+    }
+  }
+
+  /**
+   * Converts an IOExcpetion (not subclasses) to SocketException.
+   * This is typically done to indicate to upper layers that the error 
+   * was a socket error rather than often more serious exceptions like 
+   * disk errors.
+   */
+  private static IOException ioeToSocketException(IOException ioe) {
+    if (ioe.getClass().equals(IOException.class)) {
+      // "se" could be a new class in stead of SocketException.
+      IOException se = new SocketException("Original Exception : " + ioe);
+      se.initCause(ioe);
+      /* Change the stacktrace so that original trace is not truncated
+       * when printed.*/ 
+      se.setStackTrace(ioe.getStackTrace());
+      return se;
+    }
+    // otherwise just return the same exception.
+    return ioe;
+  }
+
+  /**
+   * Sends upto maxChunks chunks of data.
+   * 
+   * When blockInPosition is >= 0, assumes 'out' is a 
+   * {@link SocketOutputStream} and tries 
+   * {@link SocketOutputStream#transferToFully(FileChannel, long, int)} to
+   * send data (and updates blockInPosition).
+   */
+  private int sendChunks(ByteBuffer pkt, int maxChunks, OutputStream out) 
+                         throws IOException {
+    // Sends multiple chunks in one packet with a single write().
+
+    int len = (int) Math.min(endOffset - offset,
+                             (((long) bytesPerChecksum) * ((long) maxChunks)));
+    int numChunks = (len + bytesPerChecksum - 1)/bytesPerChecksum;
+    int packetLen = len + numChunks*checksumSize + 4;
+    boolean lastDataPacket = offset + len == endOffset && len > 0;
+    pkt.clear();
+
+
+    PacketHeader header = new PacketHeader(
+      packetLen, offset, seqno, (len == 0), len);
+    header.putInBuffer(pkt);
+
+    int checksumOff = pkt.position();
+    int checksumLen = numChunks * checksumSize;
+    byte[] buf = pkt.array();
+    
+    if (checksumSize > 0 && checksumIn != null) {
+      try {
+        checksumIn.readFully(buf, checksumOff, checksumLen);
+      } catch (IOException e) {
+        LOG.warn(" Could not read or failed to veirfy checksum for data" +
+                 " at offset " + offset + " for block " + block + " got : "
+                 + StringUtils.stringifyException(e));
+        IOUtils.closeStream(checksumIn);
+        checksumIn = null;
+        if (corruptChecksumOk) {
+          if (checksumOff < checksumLen) {
+            // Just fill the array with zeros.
+            Arrays.fill(buf, checksumOff, checksumLen, (byte) 0);
+          }
+        } else {
+          throw e;
+        }
+      }
+
+      // write in progress that we need to use to get last checksum
+      if (lastDataPacket && lastChunkChecksum != null) {
+        int start = checksumOff + checksumLen - checksumSize;
+        byte[] updatedChecksum = lastChunkChecksum.getChecksum();
+        
+        if (updatedChecksum != null) {
+          System.arraycopy(updatedChecksum, 0, buf, start, checksumSize);
+        }
+      }
+    }
+    
+    int dataOff = checksumOff + checksumLen;
+    
+    if (blockInPosition < 0) {
+      //normal transfer
+      IOUtils.readFully(blockIn, buf, dataOff, len);
+
+      if (verifyChecksum) {
+        int dOff = dataOff;
+        int cOff = checksumOff;
+        int dLeft = len;
+
+        for (int i=0; i<numChunks; i++) {
+          checksum.reset();
+          int dLen = Math.min(dLeft, bytesPerChecksum);
+          checksum.update(buf, dOff, dLen);
+          if (!checksum.compare(buf, cOff)) {
+            long failedPos = offset + len -dLeft;
+            throw new ChecksumException("Checksum failed at " + 
+                                        failedPos, failedPos);
+          }
+          dLeft -= dLen;
+          dOff += dLen;
+          cOff += checksumSize;
+        }
+      }
+      //writing is done below (mainly to handle IOException)
+    }
+    
+    try {
+      if (blockInPosition >= 0) {
+        //use transferTo(). Checks on out and blockIn are already done. 
+
+        SocketOutputStream sockOut = (SocketOutputStream)out;
+        //first write the packet
+        sockOut.write(buf, 0, dataOff);
+        // no need to flush. since we know out is not a buffered stream. 
+
+        sockOut.transferToFully(((FileInputStream)blockIn).getChannel(), 
+                                blockInPosition, len);
+
+        blockInPosition += len;
+      } else {
+        // normal transfer
+        out.write(buf, 0, dataOff + len);
+      }
+      
+    } catch (IOException e) {
+      /* exception while writing to the client (well, with transferTo(),
+       * it could also be while reading from the local file).
+       */
+      throw ioeToSocketException(e);
+    }
+
+    if (throttler != null) { // rebalancing so throttle
+      throttler.throttle(packetLen);
+    }
+
+    return len;
+  }
+
+  /**
+   * sendBlock() is used to read block and its metadata and stream the data to
+   * either a client or to another datanode. 
+   * 
+   * @param out  stream to which the block is written to
+   * @param baseStream optional. if non-null, <code>out</code> is assumed to 
+   *        be a wrapper over this stream. This enables optimizations for
+   *        sending the data, e.g. 
+   *        {@link SocketOutputStream#transferToFully(FileChannel, 
+   *        long, int)}.
+   * @param throttler for sending data.
+   * @return total bytes reads, including crc.
+   */
+  public long sendBlock(DataOutputStream out, OutputStream baseStream, 
+                 BlockTransferThrottler throttler) throws IOException {
+    if( out == null ) {
+      throw new IOException( "out stream is null" );
+    }
+    this.throttler = throttler;
+
+    long initialOffset = offset;
+    long totalRead = 0;
+    OutputStream streamForSendChunks = out;
+    
+    final long startTime = ClientTraceLog.isInfoEnabled() ? System.nanoTime() : 0;
+    try {
+      try {
+        checksum.writeHeader(out);
+        if ( chunkOffsetOK ) {
+          out.writeLong( offset );
+        }
+        out.flush();
+      } catch (IOException e) { //socket error
+        throw ioeToSocketException(e);
+      }
+      
+      int maxChunksPerPacket;
+      int pktSize = PacketHeader.PKT_HEADER_LEN;
+      
+      if (transferToAllowed && !verifyChecksum && 
+          baseStream instanceof SocketOutputStream && 
+          blockIn instanceof FileInputStream) {
+        
+        FileChannel fileChannel = ((FileInputStream)blockIn).getChannel();
+        
+        // blockInPosition also indicates sendChunks() uses transferTo.
+        blockInPosition = fileChannel.position();
+        streamForSendChunks = baseStream;
+        
+        // assure a mininum buffer size.
+        maxChunksPerPacket = (Math.max(BUFFER_SIZE, 
+                                       MIN_BUFFER_WITH_TRANSFERTO)
+                              + bytesPerChecksum - 1)/bytesPerChecksum;
+        
+        // allocate smaller buffer while using transferTo(). 
+        pktSize += checksumSize * maxChunksPerPacket;
+      } else {
+        maxChunksPerPacket = Math.max(1,
+                 (BUFFER_SIZE + bytesPerChecksum - 1)/bytesPerChecksum);
+        pktSize += (bytesPerChecksum + checksumSize) * maxChunksPerPacket;
+      }
+
+      ByteBuffer pktBuf = ByteBuffer.allocate(pktSize);
+
+      while (endOffset > offset) {
+        long len = sendChunks(pktBuf, maxChunksPerPacket, 
+                              streamForSendChunks);
+        offset += len;
+        totalRead += len + ((len + bytesPerChecksum - 1)/bytesPerChecksum*
+                            checksumSize);
+        seqno++;
+      }
+      try {
+        // send an empty packet to mark the end of the block
+        sendChunks(pktBuf, maxChunksPerPacket, streamForSendChunks);        
+        out.flush();
+      } catch (IOException e) { //socket error
+        throw ioeToSocketException(e);
+      }
+    } finally {
+      if (clientTraceFmt != null) {
+        final long endTime = System.nanoTime();
+        ClientTraceLog.info(String.format(clientTraceFmt, totalRead, initialOffset, endTime - startTime));
+      }
+      close();
+    }
+
+    blockReadFully = initialOffset == 0 && offset >= replicaVisibleLength;
+
+    return totalRead;
+  }
+  
+  boolean isBlockReadFully() {
+    return blockReadFully;
+  }
+  
+  public static interface InputStreamFactory {
+    public InputStream createStream(long offset) throws IOException; 
+  }
+  
+  private static class BlockInputStreamFactory implements InputStreamFactory {
+    private final Block block;
+    private final FSDatasetInterface data;
+
+    private BlockInputStreamFactory(Block block, FSDatasetInterface data) {
+      this.block = block;
+      this.data = data;
+    }
+
+    @Override
+    public InputStream createStream(long offset) throws IOException {
+      return data.getBlockInputStream(block, offset);
+    }
+  }
+}

Added: hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/BlockFixer.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/BlockFixer.java?rev=1026137&view=auto
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/BlockFixer.java (added)
+++ hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/BlockFixer.java Thu Oct 21 20:48:45 2010
@@ -0,0 +1,653 @@
+/**
+ * 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.hadoop.raid;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.regex.Pattern;
+import java.util.Random;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.nio.channels.SocketChannel;
+
+import org.apache.hadoop.util.DataChecksum;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
+import org.apache.hadoop.hdfs.protocol.Block;
+import org.apache.hadoop.hdfs.protocol.DataTransferProtocol;
+import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
+import org.apache.hadoop.hdfs.protocol.FSConstants;
+import org.apache.hadoop.hdfs.protocol.FSConstants.DatanodeReportType;
+import org.apache.hadoop.hdfs.protocol.LocatedBlock;
+import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
+import org.apache.hadoop.hdfs.server.common.HdfsConstants;
+import org.apache.hadoop.hdfs.server.datanode.FSDataset;
+import org.apache.hadoop.hdfs.server.datanode.RaidBlockSender;
+import org.apache.hadoop.io.Text;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.ChecksumException;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.hdfs.BlockMissingException;
+import org.apache.hadoop.hdfs.RaidDFSUtil;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.net.NetUtils;
+
+import org.apache.hadoop.raid.RaidNode;
+import org.apache.hadoop.raid.RaidUtils;
+
+/**
+ * This class fixes source file blocks using the parity file,
+ * and parity file blocks using the source file.
+ * It periodically fetches the list of corrupt files from the namenode,
+ * and figures out the location of the bad block by reading through
+ * the corrupt file.
+ */
+public class BlockFixer {
+  public static final Log LOG = LogFactory.getLog(
+                                  "org.apache.hadoop.raid.BlockFixer");
+  private java.util.HashMap<String, java.util.Date> history;
+  private int blockFixInterval = 60*1000; // 1min
+  private long numFilesFixed = 0;
+  private Configuration conf;
+  private String xorPrefix;
+  private XOREncoder xorEncoder;
+  private XORDecoder xorDecoder;
+
+  public BlockFixer(Configuration conf) throws IOException {
+    this.conf = conf;
+    history = new java.util.HashMap<String, java.util.Date>();
+    blockFixInterval = conf.getInt("raid.blockfix.interval",
+                                   blockFixInterval);
+    xorPrefix = RaidNode.getDestinationPath(conf).toUri().getPath();
+    int stripeLength = RaidNode.getStripeLength(conf);
+    xorEncoder = new XOREncoder(conf, stripeLength);
+    xorDecoder = new XORDecoder(conf, stripeLength);
+  }
+
+  public long filesFixed() {
+    return numFilesFixed;
+  }
+
+  void fixFile(Path srcPath) throws IOException {
+    if (RaidNode.isParityHarPartFile(srcPath)) {
+      processCorruptParityHarPartFile(srcPath);
+      return;
+    }
+
+    // The corrupted file is a XOR parity file
+    if (isXorParityFile(srcPath)) {
+      processCorruptParityFile(srcPath, xorEncoder);
+      return;
+    }
+
+    // The corrupted file is a source file
+
+    // Do we have a parity file for this file?
+    RaidNode.ParityFilePair ppair = null;
+    Decoder decoder = null;
+    Path destPath = null;
+    try {
+      destPath = RaidNode.getDestinationPath(conf);
+      ppair = RaidNode.getParityFile(destPath, srcPath, conf);
+      if (ppair != null) {
+        decoder = xorDecoder;
+      }
+    } catch (FileNotFoundException e) {
+    }
+    // If we have a parity file, process the file and fix it.
+    if (ppair != null) {
+      processCorruptFile(srcPath, destPath, decoder);
+    }
+  }
+
+  /**
+   * We maintain history of fixed files because a fixed file may appear in
+   * the list of corrupt files if we loop around too quickly.
+   * This function removes the old items in the history so that we can
+   * recognize files that have actually become corrupt since being fixed.
+   */
+  void purgeHistory() {
+    // Default history interval is 1 hour.
+    long historyInterval = conf.getLong(
+                             "raid.blockfix.history.interval", 3600*1000);
+    java.util.Date cutOff = new java.util.Date(
+                                   System.currentTimeMillis()-historyInterval);
+    List<String> toRemove = new java.util.ArrayList<String>();
+
+    for (String key: history.keySet()) {
+      java.util.Date item = history.get(key);
+      if (item.before(cutOff)) {
+        toRemove.add(key);
+      }
+    }
+    for (String key: toRemove) {
+      LOG.info("Removing " + key + " from history");
+      history.remove(key);
+    }
+  }
+
+  /**
+   * @return A list of corrupt files as obtained from the namenode
+   */
+  List<Path> getCorruptFiles() throws IOException {
+    DistributedFileSystem dfs = getDFS(new Path("/"));
+
+    // TODO: need an RPC here.
+    // FileStatus[] files =  dfs.getClient().namenode.getCorruptFiles();
+    FileStatus[] files = new FileStatus[0];
+    List<Path> corruptFiles = new LinkedList<Path>();
+    for (FileStatus f: files) {
+      Path p = f.getPath();
+      if (!history.containsKey(p.toString())) {
+        corruptFiles.add(p);
+      }
+    }
+    RaidUtils.filterTrash(conf, corruptFiles);
+    return corruptFiles;
+  }
+
+  /**
+   * Sorts source files ahead of parity files.
+   */
+  void sortCorruptFiles(List<Path> files) {
+    // TODO: We should first fix the files that lose more blocks
+    Comparator<Path> comp = new Comparator<Path>() {
+      public int compare(Path p1, Path p2) {
+        if (isXorParityFile(p2)) {
+          // If p2 is a parity file, p1 is smaller.
+          return -1;
+        }
+        if (isXorParityFile(p1)) {
+          // If p1 is a parity file, p2 is smaller.
+          return 1;
+        }
+        // If both are source files, they are equal.
+        return 0;
+      }
+    };
+    Collections.sort(files, comp);
+  }
+
+
+  /**
+   * Returns a DistributedFileSystem hosting the path supplied.
+   */
+  private DistributedFileSystem getDFS(Path p) throws IOException {
+    return (DistributedFileSystem) p.getFileSystem(conf);
+  }
+
+  /**
+   * Reads through a corrupt source file fixing corrupt blocks on the way.
+   * @param srcPath Path identifying the corrupt file.
+   * @throws IOException
+   */
+  void processCorruptFile(Path srcPath, Path destPath, Decoder decoder)
+      throws IOException {
+    LOG.info("Processing corrupt file " + srcPath);
+
+    DistributedFileSystem srcFs = getDFS(srcPath);
+    FileStatus srcStat = srcFs.getFileStatus(srcPath);
+    long blockSize = srcStat.getBlockSize();
+    long srcFileSize = srcStat.getLen();
+    String uriPath = srcPath.toUri().getPath();
+
+    int numBlocksFixed = 0;
+    List<LocatedBlock> corrupt =
+      RaidDFSUtil.corruptBlocksInFile(srcFs, uriPath, 0, srcFileSize);
+    for (LocatedBlock lb: corrupt) {
+      Block corruptBlock = lb.getBlock();
+      long corruptOffset = lb.getStartOffset();
+
+      LOG.info("Found corrupt block " + corruptBlock +
+          ", offset " + corruptOffset);
+
+      final long blockContentsSize =
+        Math.min(blockSize, srcFileSize - corruptOffset);
+      File localBlockFile =
+        File.createTempFile(corruptBlock.getBlockName(), ".tmp");
+      localBlockFile.deleteOnExit();
+
+      try {
+        RaidNode.ParityFilePair parityPair = RaidNode.getParityFile(
+            destPath, srcPath, conf);
+
+        decoder.recoverBlockToFile(srcFs, srcPath, parityPair.getFileSystem(),
+          parityPair.getPath(), blockSize, corruptOffset, localBlockFile,
+          blockContentsSize);
+
+        // We have a the contents of the block, send them.
+        DatanodeInfo datanode = chooseDatanode(lb.getLocations());
+        computeMetdataAndSendFixedBlock(
+          datanode, localBlockFile, lb, blockContentsSize);
+        numBlocksFixed++;
+
+        LOG.info("Adding " + srcPath + " to history");
+        history.put(srcPath.toString(), new java.util.Date());
+      } finally {
+        localBlockFile.delete();
+      }
+    }
+    LOG.info("Fixed " + numBlocksFixed + " blocks in " + srcPath);
+    numFilesFixed++;
+  }
+
+  boolean isXorParityFile(Path p) {
+    String pathStr = p.toUri().getPath();
+    if (pathStr.contains(RaidNode.HAR_SUFFIX)) {
+      return false;
+    }
+    return pathStr.startsWith(xorPrefix);
+  }
+
+  /**
+   * Reads through a parity file, fixing corrupt blocks on the way.
+   * This function uses the corresponding source file to regenerate parity
+   * file blocks.
+   */
+  void processCorruptParityFile(Path parityPath, Encoder encoder)
+      throws IOException {
+    LOG.info("Processing corrupt file " + parityPath);
+    Path srcPath = sourcePathFromParityPath(parityPath);
+    if (srcPath == null) {
+      LOG.warn("Unusable parity file " + parityPath);
+      return;
+    }
+
+    DistributedFileSystem parityFs = getDFS(parityPath);
+    FileStatus parityStat = parityFs.getFileStatus(parityPath);
+    long blockSize = parityStat.getBlockSize();
+    long parityFileSize = parityStat.getLen();
+    FileStatus srcStat = getDFS(srcPath).getFileStatus(srcPath);
+    long srcFileSize = srcStat.getLen();
+
+    // Check timestamp.
+    if (srcStat.getModificationTime() != parityStat.getModificationTime()) {
+      LOG.info("Mismatching timestamp for " + srcPath + " and " + parityPath +
+               ", moving on...");
+      return;
+    }
+
+    String uriPath = parityPath.toUri().getPath();
+    int numBlocksFixed = 0;
+    List<LocatedBlock> corrupt = RaidDFSUtil.corruptBlocksInFile(
+      parityFs, uriPath, 0, parityFileSize);
+    for (LocatedBlock lb: corrupt) {
+      Block corruptBlock = lb.getBlock();
+      long corruptOffset = lb.getStartOffset();
+
+      LOG.info("Found corrupt block " + corruptBlock +
+          ", offset " + corruptOffset);
+
+      File localBlockFile =
+        File.createTempFile(corruptBlock.getBlockName(), ".tmp");
+      localBlockFile.deleteOnExit();
+
+      try {
+        encoder.recoverParityBlockToFile(parityFs, srcPath, srcFileSize,
+            blockSize, parityPath, corruptOffset, localBlockFile);
+        // We have a the contents of the block, send them.
+        DatanodeInfo datanode = chooseDatanode(lb.getLocations());
+        computeMetdataAndSendFixedBlock(
+          datanode, localBlockFile, lb, blockSize);
+
+        numBlocksFixed++;
+        LOG.info("Adding " + parityPath + " to history");
+        history.put(parityPath.toString(), new java.util.Date());
+      } finally {
+        localBlockFile.delete();
+      }
+    }
+    LOG.info("Fixed " + numBlocksFixed + " blocks in " + parityPath);
+    numFilesFixed++;
+  }
+
+  /**
+   * Reads through a parity HAR part file, fixing corrupt blocks on the way.
+   * A HAR block can contain many file blocks, as long as the HAR part file
+   * block size is a multiple of the file block size.
+   */
+  void processCorruptParityHarPartFile(Path partFile) throws IOException {
+    LOG.info("Processing corrupt file " + partFile);
+    // Get some basic information.
+    DistributedFileSystem dfs = getDFS(partFile);
+    FileStatus partFileStat = dfs.getFileStatus(partFile);
+    long partFileSize = partFileStat.getLen();
+    long partFileBlockSize = partFileStat.getBlockSize();
+    LOG.info(partFile + " has block size " + partFileBlockSize);
+
+    // Find the path to the index file.
+    // Parity file HARs are only one level deep, so the index files is at the
+    // same level as the part file.
+    String harDirectory = partFile.toUri().getPath(); // Temporarily.
+    harDirectory =
+      harDirectory.substring(0, harDirectory.lastIndexOf(Path.SEPARATOR));
+    Path indexFile = new Path(harDirectory + "/" + HarIndex.indexFileName);
+    FileStatus indexStat = dfs.getFileStatus(indexFile);
+    // Parses through the HAR index file.
+    HarIndex harIndex = new HarIndex(dfs.open(indexFile), indexStat.getLen());
+
+    String uriPath = partFile.toUri().getPath();
+    int numBlocksFixed = 0;
+    List<LocatedBlock> corrupt = RaidDFSUtil.corruptBlocksInFile(
+      dfs, uriPath, 0, partFileSize);
+    for (LocatedBlock lb: corrupt) {
+      Block corruptBlock = lb.getBlock();
+      long corruptOffset = lb.getStartOffset();
+
+      File localBlockFile =
+        File.createTempFile(corruptBlock.getBlockName(), ".tmp");
+      localBlockFile.deleteOnExit();
+      processCorruptParityHarPartBlock(
+        dfs, partFile, corruptBlock, corruptOffset, partFileStat, harIndex,
+        localBlockFile);
+      // Now we have recovered the part file block locally, send it.
+      try {
+        DatanodeInfo datanode = chooseDatanode(lb.getLocations());
+        computeMetdataAndSendFixedBlock(datanode, localBlockFile,
+          lb, localBlockFile.length());
+        numBlocksFixed++;
+
+        LOG.info("Adding " + partFile + " to history");
+        history.put(partFile.toString(), new java.util.Date());
+      } finally {
+        localBlockFile.delete();
+      }
+    }
+    LOG.info("Fixed " + numBlocksFixed + " blocks in " + partFile);
+    numFilesFixed++;
+  }
+
+  /**
+   * This fixes a single part file block by recovering in sequence each
+   * parity block in the part file block.
+   */
+  private void processCorruptParityHarPartBlock(
+    FileSystem dfs, Path partFile, Block corruptBlock, long corruptOffset,
+    FileStatus partFileStat, HarIndex harIndex, File localBlockFile)
+    throws IOException {
+    String partName = partFile.toUri().getPath(); // Temporarily.
+    partName = partName.substring(1 + partName.lastIndexOf(Path.SEPARATOR));
+
+    OutputStream out = new FileOutputStream(localBlockFile);
+
+    try {
+      // A HAR part file block could map to several parity files. We need to
+      // use all of them to recover this block.
+      final long corruptEnd = Math.min(corruptOffset + partFileStat.getBlockSize(),
+                                      partFileStat.getLen());
+      for (long offset = corruptOffset; offset < corruptEnd; ) {
+        HarIndex.IndexEntry entry = harIndex.findEntry(partName, offset);
+        if (entry == null) {
+          String msg = "Corrupt index file has no matching index entry for " +
+            partName + ":" + offset;
+          LOG.warn(msg);
+          throw new IOException(msg);
+        }
+        Path parityFile = new Path(entry.fileName);
+        Encoder encoder;
+        if (isXorParityFile(parityFile)) {
+          encoder = xorEncoder;
+        } else {
+          String msg = "Could not figure out parity file correctly";
+          LOG.warn(msg);
+          throw new IOException(msg);
+        }
+        Path srcFile = sourcePathFromParityPath(parityFile);
+        FileStatus srcStat = dfs.getFileStatus(srcFile);
+        if (srcStat.getModificationTime() != entry.mtime) {
+          String msg = "Modification times of " + parityFile + " and " + srcFile +
+            " do not match.";
+          LOG.warn(msg);
+          throw new IOException(msg);
+        }
+        long corruptOffsetInParity = offset - entry.startOffset;
+        LOG.info(partFile + ":" + offset + " maps to " +
+                 parityFile + ":" + corruptOffsetInParity +
+                 " and will be recovered from " + srcFile);
+        encoder.recoverParityBlockToStream(dfs, srcFile, srcStat.getLen(),
+          srcStat.getBlockSize(), parityFile, corruptOffsetInParity, out);
+        // Finished recovery of one parity block. Since a parity block has the
+        // same size as a source block, we can move offset by source block size.
+        offset += srcStat.getBlockSize();
+        LOG.info("Recovered " + srcStat.getBlockSize() + " part file bytes ");
+        if (offset > corruptEnd) {
+          String msg =
+            "Recovered block spills across part file blocks. Cannot continue...";
+          throw new IOException(msg);
+        }
+      }
+    } finally {
+      out.close();
+    }
+  }
+
+  /**
+   * Choose a datanode (hostname:portnumber). The datanode is chosen at
+   * random from the live datanodes.
+   * @param locationsToAvoid locations to avoid.
+   * @return A string in the format name:port.
+   * @throws IOException
+   */
+  private DatanodeInfo chooseDatanode(DatanodeInfo[] locationsToAvoid)
+    throws IOException {
+    DistributedFileSystem dfs = getDFS(new Path("/"));
+    DatanodeInfo[] live = dfs.getClient().datanodeReport(
+                                                 DatanodeReportType.LIVE);
+    LOG.info("Choosing a datanode from " + live.length +
+      " live nodes while avoiding " + locationsToAvoid.length);
+    Random rand = new Random();
+    DatanodeInfo chosen = null;
+    int maxAttempts = 1000;
+    for (int i = 0; i < maxAttempts && chosen == null; i++) {
+      int idx = rand.nextInt(live.length);
+      chosen = live[idx];
+      for (DatanodeInfo avoid: locationsToAvoid) {
+        if (chosen.name.equals(avoid.name)) {
+          LOG.info("Avoiding " + avoid.name);
+          chosen = null;
+          break;
+        }
+      }
+    }
+    if (chosen == null) {
+      throw new IOException("Could not choose datanode");
+    }
+    LOG.info("Choosing datanode " + chosen.name);
+    return chosen;
+  }
+
+  /**
+   * Reads data from the data stream provided and computes metadata.
+   */
+  static DataInputStream computeMetadata(
+    Configuration conf, InputStream dataStream) throws IOException {
+    ByteArrayOutputStream mdOutBase = new ByteArrayOutputStream(1024*1024);
+    DataOutputStream mdOut = new DataOutputStream(mdOutBase);
+
+    // First, write out the version.
+    mdOut.writeShort(FSDataset.METADATA_VERSION);
+
+    // Create a summer and write out its header.
+    int bytesPerChecksum = conf.getInt("io.bytes.per.checksum", 512);
+    DataChecksum sum = DataChecksum.newDataChecksum(
+                        DataChecksum.CHECKSUM_CRC32,
+                        bytesPerChecksum);
+    sum.writeHeader(mdOut);
+
+    // Buffer to read in a chunk of data.
+    byte[] buf = new byte[bytesPerChecksum];
+    // Buffer to store the checksum bytes.
+    byte[] chk = new byte[sum.getChecksumSize()];
+
+    // Read data till we reach the end of the input stream.
+    int bytesSinceFlush = 0;
+    while (true) {
+      // Read some bytes.
+      int bytesRead = dataStream.read(
+        buf, bytesSinceFlush, bytesPerChecksum-bytesSinceFlush);
+      if (bytesRead == -1) {
+        if (bytesSinceFlush > 0) {
+          boolean reset = true;
+          sum.writeValue(chk, 0, reset); // This also resets the sum.
+          // Write the checksum to the stream.
+          mdOut.write(chk, 0, chk.length);
+          bytesSinceFlush = 0;
+        }
+        break;
+      }
+      // Update the checksum.
+      sum.update(buf, bytesSinceFlush, bytesRead);
+      bytesSinceFlush += bytesRead;
+
+      // Flush the checksum if necessary.
+      if (bytesSinceFlush == bytesPerChecksum) {
+        boolean reset = true;
+        sum.writeValue(chk, 0, reset); // This also resets the sum.
+        // Write the checksum to the stream.
+        mdOut.write(chk, 0, chk.length);
+        bytesSinceFlush = 0;
+      }
+    }
+
+    byte[] mdBytes = mdOutBase.toByteArray();
+    return new DataInputStream(new ByteArrayInputStream(mdBytes));
+  }
+
+  private void computeMetdataAndSendFixedBlock(
+    DatanodeInfo datanode,
+    File localBlockFile, LocatedBlock block, long blockSize
+    ) throws IOException {
+
+    LOG.info("Computing metdata");
+    InputStream blockContents = null;
+    DataInputStream blockMetadata = null;
+    try {
+      blockContents = new FileInputStream(localBlockFile);
+      blockMetadata = computeMetadata(conf, blockContents);
+      blockContents.close();
+      // Reopen
+      blockContents = new FileInputStream(localBlockFile);
+      sendFixedBlock(datanode, blockContents, blockMetadata, block, blockSize);
+    } finally {
+      if (blockContents != null) {
+        blockContents.close();
+        blockContents = null;
+      }
+      if (blockMetadata != null) {
+        blockMetadata.close();
+        blockMetadata = null;
+      }
+    }
+  }
+
+  /**
+   * Send a generated block to a datanode.
+   * @param datanode Chosen datanode name in host:port form.
+   * @param blockContents Stream with the block contents.
+   * @param corruptBlock Block identifying the block to be sent.
+   * @param blockSize size of the block.
+   * @throws IOException
+   */
+  private void sendFixedBlock(
+    DatanodeInfo datanode,
+    final InputStream blockContents, DataInputStream metadataIn,
+    LocatedBlock block, long blockSize
+    ) throws IOException {
+    InetSocketAddress target = NetUtils.createSocketAddr(datanode.name);
+    Socket sock = SocketChannel.open().socket();
+
+    int readTimeout =  conf.getInt(
+        "raid.blockfix.read.timeout", HdfsConstants.READ_TIMEOUT);
+    NetUtils.connect(sock, target, readTimeout);
+    sock.setSoTimeout(readTimeout);
+
+    int writeTimeout = conf.getInt(
+        "raid.blockfix.write.timeout", HdfsConstants.WRITE_TIMEOUT);
+
+    OutputStream baseStream = NetUtils.getOutputStream(sock, writeTimeout);
+    DataOutputStream out = new DataOutputStream(
+        new BufferedOutputStream(baseStream, FSConstants.SMALL_BUFFER_SIZE));
+
+    boolean corruptChecksumOk = false;
+    boolean chunkOffsetOK = false;
+    boolean verifyChecksum = true;
+    boolean transferToAllowed = false;
+
+    try {
+      LOG.info("Sending block " + block.getBlock() +
+          " from " + sock.getLocalSocketAddress().toString() +
+          " to " + sock.getRemoteSocketAddress().toString() +
+          " " + blockSize + " bytes");
+      RaidBlockSender blockSender = new RaidBlockSender(
+          block.getBlock(), blockSize, 0, blockSize,
+          corruptChecksumOk, chunkOffsetOK, verifyChecksum, transferToAllowed,
+          metadataIn, new RaidBlockSender.InputStreamFactory() {
+          @Override
+          public InputStream createStream(long offset) throws IOException {
+            // we are passing 0 as the offset above, so we can safely ignore
+            // the offset passed
+            return blockContents;
+          }
+        });
+
+      DatanodeInfo[] nodes = new DatanodeInfo[]{datanode};
+      DataTransferProtocol.Sender.opWriteBlock(
+        out, block.getBlock(), 1,
+        DataTransferProtocol.BlockConstructionStage.PIPELINE_SETUP_CREATE,
+        0, blockSize, 0, "", null, nodes, block.getBlockToken());
+      blockSender.sendBlock(out, baseStream, null);
+
+      LOG.info("Sent block " + block.getBlock() + " to " + datanode.name);
+    } finally {
+      out.close();
+    }
+  }
+
+  Path sourcePathFromParityPath(Path parityPath) {
+    String parityPathStr = parityPath.toUri().getPath();
+    if (parityPathStr.startsWith(xorPrefix)) {
+      // Remove the prefix to get the source file.
+      String src = parityPathStr.replaceFirst(xorPrefix, "");
+      return new Path(src);
+    }
+    return null;
+  }
+}
+

Added: hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/HarIndex.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/HarIndex.java?rev=1026137&view=auto
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/HarIndex.java (added)
+++ hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/HarIndex.java Thu Oct 21 20:48:45 2010
@@ -0,0 +1,144 @@
+/**
+ * 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.hadoop.raid;
+
+import java.io.InputStream;
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.net.URLDecoder;
+import java.util.List;
+import java.util.LinkedList;
+
+import org.apache.hadoop.util.LineReader;
+import org.apache.hadoop.io.Text;
+
+/**
+ * Represents the contents of a HAR Index file. The HAR is assumed to be
+ * comprising of RAID parity files only and no directories.
+ */
+public class HarIndex {
+  public static final String indexFileName = "_index";
+  private List<IndexEntry> entries = new LinkedList<IndexEntry>();
+
+  /**
+   * Represents information in a single line of the HAR index file.
+   */
+  public static class IndexEntry {
+    String fileName; // Name of the file in the part file.
+    long startOffset; // Start offset within the part file.
+    long length; // Length of this file within the part file.
+    long mtime; // Modification time of the file.
+    String partFileName; // Name of the part file.
+
+    IndexEntry(String fileName, long startOffset, long length,
+                long mtime, String partFileName) {
+      this.fileName = fileName;
+      this.startOffset = startOffset;
+      this.length = length;
+      this.mtime = mtime;
+      this.partFileName = partFileName;
+    }
+
+    public String toString() {
+      return "fileName=" + fileName +
+             ", startOffset=" + startOffset +
+             ", length=" + length +
+             ", mtime=" + mtime +
+             ", partFileName=" + partFileName;
+    }
+  }
+
+  /**
+   * Constructor that reads the contents of the index file.
+   * @param in An input stream to the index file.
+   * @param max The size of the index file.
+   * @throws IOException
+   */
+  public HarIndex(InputStream in, long max) throws IOException {
+    LineReader lineReader = new LineReader(in);
+    Text text = new Text();
+    long nread = 0;
+    while (nread < max) {
+      int n = lineReader.readLine(text);
+      nread += n;
+      String line = text.toString();
+      try {
+        parseLine(line);
+      } catch (UnsupportedEncodingException e) {
+        throw new IOException("UnsupportedEncodingException after reading " +
+                              nread + "bytes");
+      }
+    }
+  }
+
+  /**
+   * Parses each line and extracts relevant information.
+   * @param line
+   * @throws UnsupportedEncodingException
+   */
+  void parseLine(String line) throws UnsupportedEncodingException {
+    String[] splits = line.split(" ");
+
+    boolean isDir = "dir".equals(splits[1]) ? true: false;
+    if (!isDir && splits.length >= 6) {
+      String name = URLDecoder.decode(splits[0], "UTF-8");
+      String partName = URLDecoder.decode(splits[2], "UTF-8");
+      long startIndex = Long.parseLong(splits[3]);
+      long length = Long.parseLong(splits[4]);
+      String[] newsplits = URLDecoder.decode(splits[5],"UTF-8").split(" ");
+      if (newsplits != null && newsplits.length >= 5) {
+        long mtime = Long.parseLong(newsplits[0]);
+        IndexEntry entry = new IndexEntry(
+          name, startIndex, length, mtime, partName);
+        entries.add(entry);
+      }
+    }
+  }
+
+  /**
+   * Finds the index entry corresponding to a HAR partFile at an offset.
+   * @param partName The name of the part file (part-*).
+   * @param partFileOffset The offset into the part file.
+   * @return The entry corresponding to partName:partFileOffset.
+   */
+  public IndexEntry findEntry(String partName, long partFileOffset) {
+    for (IndexEntry e: entries) {
+      boolean nameMatch = partName.equals(e.partFileName);
+      boolean inRange = (partFileOffset >= e.startOffset) &&
+                        (partFileOffset < e.startOffset + e.length);
+      if (nameMatch && inRange) {
+        return e;
+      }
+    }
+    return null;
+  }
+
+  /**
+   * Finds the index entry corresponding to a file in the archive
+   */
+  public IndexEntry findEntryByFileName(String fileName) {
+    for (IndexEntry e: entries) {
+      if (fileName.equals(e.fileName)) {
+        return e;
+      }
+    }
+    return null;
+  }
+
+}

Modified: hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidNode.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidNode.java?rev=1026137&r1=1026136&r2=1026137&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidNode.java (original)
+++ hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidNode.java Thu Oct 21 20:48:45 2010
@@ -30,6 +30,8 @@ import java.util.Map;
 import java.util.Random;
 import java.util.Set;
 import java.util.HashSet;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 import java.lang.Thread;
 import java.net.InetSocketAddress;
 import java.net.URI;
@@ -79,6 +81,9 @@ public class RaidNode implements RaidPro
   public static final String DEFAULT_RAID_LOCATION = "/raid";
   public static final String RAID_LOCATION_KEY = "hdfs.raid.locations";
   public static final String HAR_SUFFIX = "_raid.har";
+  public static final Pattern PARITY_HAR_PARTFILE_PATTERN =
+    Pattern.compile(".*" + HAR_SUFFIX + "/part-.*");
+
   
   /** RPC server */
   private Server server;
@@ -1149,6 +1154,11 @@ public class RaidNode implements RaidPro
     return conf.getInt(STRIPE_LENGTH_KEY, DEFAULT_STRIPE_LENGTH);
   }
 
+  static boolean isParityHarPartFile(Path p) {
+    Matcher m = PARITY_HAR_PARTFILE_PATTERN.matcher(p.toUri().getPath());
+    return m.matches();
+  }
+
   /**
    * Returns current time.
    */

Modified: hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidShell.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidShell.java?rev=1026137&r1=1026136&r2=1026137&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidShell.java (original)
+++ hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidShell.java Thu Oct 21 20:48:45 2010
@@ -48,9 +48,13 @@ import org.apache.hadoop.raid.protocol.R
  * A {@link RaidShell} that allows browsing configured raid policies.
  */
 public class RaidShell extends Configured implements Tool {
+  static {
+    Configuration.addDefaultResource("hdfs-default.xml");
+    Configuration.addDefaultResource("hdfs-site.xml");
+  }
   public static final Log LOG = LogFactory.getLog( "org.apache.hadoop.RaidShell");
   public RaidProtocol raidnode;
-  final RaidProtocol rpcRaidnode;
+  RaidProtocol rpcRaidnode;
   private UserGroupInformation ugi;
   volatile boolean clientRunning = true;
   private Configuration conf;
@@ -62,28 +66,21 @@ public class RaidShell extends Configure
    * configuration options.
    * @throws IOException
    */
-  public RaidShell() throws IOException {
-    this(new Configuration());
-  }
-
-  /**
-   * The RaidShell connects to the specified RaidNode and performs basic
-   * configuration options.
-   * @param conf The Hadoop configuration
-   * @throws IOException
-   */
   public RaidShell(Configuration conf) throws IOException {
-    this(conf, RaidNode.getAddress(conf));
+    super(conf);
+    this.conf = conf;
   }
 
-  public RaidShell(Configuration conf, InetSocketAddress address) throws IOException {
-    super(conf);
+  void initializeRpc(Configuration conf, InetSocketAddress address) throws IOException {
     this.ugi = UserGroupInformation.getCurrentUser();
-
     this.rpcRaidnode = createRPCRaidnode(address, conf, ugi);
     this.raidnode = createRaidnode(rpcRaidnode);
   }
-  
+
+  void initializeLocal(Configuration conf) throws IOException {
+    this.ugi = UserGroupInformation.getCurrentUser();
+  }
+
   public static RaidProtocol createRaidnode(Configuration conf) throws IOException {
     return createRaidnode(RaidNode.getAddress(conf), conf);
   }
@@ -92,7 +89,6 @@ public class RaidShell extends Configure
       Configuration conf) throws IOException {
     return createRaidnode(createRPCRaidnode(raidNodeAddr, conf,
       UserGroupInformation.getCurrentUser()));
-
   }
 
   private static RaidProtocol createRPCRaidnode(InetSocketAddress raidNodeAddr,
@@ -153,13 +149,17 @@ public class RaidShell extends Configure
       System.err.println("Usage: java RaidShell" + 
                          " [-showConfig]"); 
     } else if ("-recover".equals(cmd)) {
-      System.err.println("Usage: java CronShell" +
+      System.err.println("Usage: java RaidShell" +
                          " [-recover srcPath1 corruptOffset]");
+    } else if ("-recoverBlocks".equals(cmd)) {
+      System.err.println("Usage: java RaidShell" +
+                         " [-recoverBlocks path1 path2...]");
     } else {
       System.err.println("Usage: java RaidShell");
       System.err.println("           [-showConfig ]");
       System.err.println("           [-help [cmd]]");
       System.err.println("           [-recover srcPath1 corruptOffset]");
+      System.err.println("           [-recoverBlocks path1 path2...]");
       System.err.println();
       ToolRunner.printGenericCommandUsage(System.err);
     }
@@ -195,9 +195,15 @@ public class RaidShell extends Configure
 
     try {
       if ("-showConfig".equals(cmd)) {
+        initializeRpc(conf, RaidNode.getAddress(conf));
         exitCode = showConfig(cmd, argv, i);
       } else if ("-recover".equals(cmd)) {
+        initializeRpc(conf, RaidNode.getAddress(conf));
         exitCode = recoverAndPrint(cmd, argv, i);
+      } else if ("-recoverBlocks".equals(cmd)) {
+        initializeLocal(conf);
+        recoverBlocks(argv, i);
+        exitCode = 0;
       } else {
         exitCode = -1;
         System.err.println(cmd.substring(1) + ": Unknown command");
@@ -278,29 +284,37 @@ public class RaidShell extends Configure
     }
     return exitCode;
   }
-  
+
+  public void recoverBlocks(String[] args, int startIndex)
+    throws IOException {
+    LOG.info("Recovering blocks for " + (args.length - startIndex) + " files");
+    BlockFixer fixer = new BlockFixer(conf);
+    for (int i = startIndex; i < args.length; i++) {
+      String path = args[i];
+      fixer.fixFile(new Path(path));
+    }
+  }
+
   /**
    * main() has some simple utility methods
    */
   public static void main(String argv[]) throws Exception {
     RaidShell shell = null;
     try {
-      shell = new RaidShell();
+      shell = new RaidShell(new Configuration());
+      int res = ToolRunner.run(shell, argv);
+      System.exit(res);
     } catch (RPC.VersionMismatch v) {
       System.err.println("Version Mismatch between client and server" +
                          "... command aborted.");
       System.exit(-1);
     } catch (IOException e) {
-      System.err.println("Bad connection to RaidNode. command aborted.");
+      System.err.
+        println("Bad connection to RaidNode or NameNode. command aborted.");
+      System.err.println(e.getMessage());
       System.exit(-1);
-    }
-
-    int res;
-    try {
-      res = ToolRunner.run(shell, argv);
     } finally {
       shell.close();
     }
-    System.exit(res);
   }
 }

Modified: hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidUtils.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidUtils.java?rev=1026137&r1=1026136&r2=1026137&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidUtils.java (original)
+++ hadoop/mapreduce/trunk/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidUtils.java Thu Oct 21 20:48:45 2010
@@ -22,12 +22,32 @@ import java.io.InputStream;
 import java.io.OutputStream;
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.regex.Pattern;
 
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.PositionedReadable;
 import org.apache.hadoop.fs.Seekable;
 import org.apache.hadoop.io.IOUtils;
 
 public class RaidUtils {
+  /**
+   * Removes files matching the trash file pattern.
+   */
+  public static void filterTrash(Configuration conf, List<Path> files) {
+    // Remove files under Trash.
+    String trashPattern = conf.get("raid.blockfixer.trash.pattern",
+                                   "^/user/.*/\\.Trash.*");
+    for (Iterator<Path> it = files.iterator(); it.hasNext(); ) {
+      String pathStr = it.next().toString();
+      if (Pattern.matches(trashPattern, pathStr)) {
+        it.remove();
+      }
+    }
+  }
+
   public static void readTillEnd(InputStream in, byte[] buf, boolean eofOK)
     throws IOException {
     int toRead = buf.length;

Modified: hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/hdfs/TestRaidDfs.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/hdfs/TestRaidDfs.java?rev=1026137&r1=1026136&r2=1026137&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/hdfs/TestRaidDfs.java (original)
+++ hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/hdfs/TestRaidDfs.java Thu Oct 21 20:48:45 2010
@@ -142,7 +142,8 @@ public class TestRaidDfs extends TestCas
   private LocatedBlocks getBlockLocations(Path file, long length)
     throws IOException {
     DistributedFileSystem dfs = (DistributedFileSystem) fileSys;
-    return dfs.getClient().namenode.getBlockLocations(file.toString(), 0, length);
+    return RaidDFSUtil.getBlockLocations(
+      dfs, file.toUri().getPath(), 0, length);
   }
 
   private LocatedBlocks getBlockLocations(Path file)
@@ -190,8 +191,8 @@ public class TestRaidDfs extends TestCas
     while (true) {
       LocatedBlocks locations = null;
       DistributedFileSystem dfs = (DistributedFileSystem) fileSys;
-      locations = dfs.getClient().namenode.getBlockLocations(
-                    file.toString(), 0, parityStat.getLen());
+      locations = RaidDFSUtil.getBlockLocations(
+        dfs, file.toUri().getPath(), 0, parityStat.getLen());
       if (!locations.isUnderConstruction()) {
         break;
       }

Added: hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/raid/TestHarIndexParser.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/raid/TestHarIndexParser.java?rev=1026137&view=auto
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/raid/TestHarIndexParser.java (added)
+++ hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/raid/TestHarIndexParser.java Thu Oct 21 20:48:45 2010
@@ -0,0 +1,72 @@
+/**
+ * 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.hadoop.raid;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStreamWriter;
+import java.io.UnsupportedEncodingException;
+import java.nio.charset.Charset;
+
+import junit.framework.TestCase;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+public class TestHarIndexParser extends TestCase {
+  final static Log LOG = LogFactory.getLog(TestHarIndexParser.class);
+  File indexFile = null;
+
+  protected void setUp() throws FileNotFoundException, IOException {
+    LOG.info("TestHarIndexParser.setUp()");
+    indexFile = File.createTempFile("harindex", ".tmp");
+    indexFile.deleteOnExit();
+    OutputStreamWriter out = new OutputStreamWriter(
+      new FileOutputStream(indexFile),
+      Charset.forName("UTF-8"));
+    out.write("%2F dir 1282018162460+0+493+hadoop+hadoop 0 0 f1 f2 f3 f4\n");
+    out.write("%2Ff1 file part-0 0 1024 1282018141145+1282018140822+420+hadoop+hadoop\n");
+    out.write("%2Ff3 file part-0 2048 1024 1282018148590+1282018148255+420+hadoop+hadoop\n");
+    out.write("%2Ff2 file part-0 1024 1024 1282018144198+1282018143852+420+hadoop+hadoop\n");
+    out.write("%2Ff4 file part-1 0 1024000 1282018162959+1282018162460+420+hadoop+hadoop\n");
+    out.flush();
+    out.close();
+  }
+
+  protected void tearDown() {
+    LOG.info("TestHarIndexParser.tearDown()");
+    if (indexFile != null)
+      indexFile.delete();
+  }
+
+  public void testHarIndexParser()
+    throws UnsupportedEncodingException, IOException {
+    LOG.info("testHarIndexParser started.");
+    InputStream in = new FileInputStream(indexFile);
+    long size = indexFile.length();
+    HarIndex parser = new HarIndex(in, size);
+
+    HarIndex.IndexEntry entry = parser.findEntry("part-0", 2100);
+    assertEquals("/f3", entry.fileName);
+
+    LOG.info("testHarIndexParser finished.");
+  }
+}

Modified: hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidNode.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidNode.java?rev=1026137&r1=1026136&r2=1026137&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidNode.java (original)
+++ hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidNode.java Thu Oct 21 20:48:45 2010
@@ -278,18 +278,6 @@ public class TestRaidNode extends TestCa
       cnode = RaidNode.createRaidNode(null, conf);
       int times = 10;
 
-      while (times-- > 0) {
-        try {
-          shell = new RaidShell(conf, cnode.getListenerAddress());
-        } catch (Exception e) {
-          LOG.info("doTestPathFilter unable to connect to " + 
-              cnode.getListenerAddress() + " retrying....");
-          Thread.sleep(1000);
-          continue;
-        }
-        break;
-      }
-      LOG.info("doTestPathFilter created RaidShell.");
       FileStatus[] listPaths = null;
 
       // wait till file is raided
@@ -322,6 +310,8 @@ public class TestRaidNode extends TestCa
       Thread.sleep(20000); // Without this wait, unit test crashes
 
       // check for error at beginning of file
+      shell = new RaidShell(conf);
+      shell.initializeRpc(conf, cnode.getListenerAddress());
       if (numBlock >= 1) {
         LOG.info("doTestPathFilter Check error at beginning of file.");
         simulateError(shell, fileSys, file1, crc1, 0);

Added: hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidShell.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidShell.java?rev=1026137&view=auto
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidShell.java (added)
+++ hadoop/mapreduce/trunk/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidShell.java Thu Oct 21 20:48:45 2010
@@ -0,0 +1,307 @@
+/**
+ * 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.hadoop.raid;
+
+import java.io.File;
+import java.io.FileWriter;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.PrintWriter;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.GregorianCalendar;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Properties;
+import java.util.Random;
+import java.util.zip.CRC32;
+
+import junit.framework.TestCase;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.hadoop.fs.BlockLocation;
+import org.apache.hadoop.fs.ChecksumException;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FilterFileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.protocol.ClientProtocol;
+import org.apache.hadoop.hdfs.protocol.LocatedBlock;
+import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
+import org.apache.hadoop.hdfs.protocol.Block;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
+import org.apache.hadoop.hdfs.DistributedRaidFileSystem;
+import org.apache.hadoop.hdfs.TestRaidDfs;
+import org.apache.hadoop.hdfs.RaidDFSUtil;
+import org.apache.hadoop.raid.RaidNode;
+
+
+public class TestRaidShell extends TestCase {
+  final static Log LOG = LogFactory.getLog(
+                            "org.apache.hadoop.raid.TestRaidShell");
+  final static String TEST_DIR = new File(System.getProperty("test.build.data",
+      "build/contrib/raid/test/data")).getAbsolutePath();
+  final static String CONFIG_FILE = new File(TEST_DIR,
+      "test-raid.xml").getAbsolutePath();
+  final static long RELOAD_INTERVAL = 1000;
+  final static int NUM_DATANODES = 3;
+  Configuration conf;
+  String namenode = null;
+  MiniDFSCluster dfs = null;
+  String hftp = null;
+  FileSystem fileSys = null;
+  RaidNode cnode = null;
+  Random rand = new Random();
+
+  /**
+   * Create a file with three stripes, corrupt a block each in two stripes,
+   * and wait for the the file to be fixed.
+   */
+  public void testBlockFix() throws Exception {
+    LOG.info("Test testBlockFix started.");
+    long blockSize = 8192L;
+    int stripeLength = 3;
+    mySetup(stripeLength, -1);
+    Path file1 = new Path("/user/dhruba/raidtest/file1");
+    Path destPath = new Path("/destraid/user/dhruba/raidtest");
+    Path parityFile = new Path(destPath, "file1");
+    long crc1 = TestRaidDfs.createTestFilePartialLastBlock(fileSys, file1,
+                                                          1, 7, blockSize);
+    long file1Len = fileSys.getFileStatus(file1).getLen();
+    LOG.info("Test testBlockFix created test files");
+
+    // create an instance of the RaidNode
+    Configuration localConf = new Configuration(conf);
+    localConf.set(RaidNode.RAID_LOCATION_KEY, "/destraid");
+    localConf.setInt("raid.blockfix.interval", 1000);
+    // the RaidNode does the raiding inline (instead of submitting to map/reduce)
+    conf.setBoolean("fs.raidnode.local", true);
+    cnode = RaidNode.createRaidNode(null, localConf);
+
+    try {
+      TestRaidDfs.waitForFileRaided(LOG, fileSys, file1, destPath);
+      cnode.stop();
+      cnode.join();
+      cnode = null;
+
+      FileStatus srcStat = fileSys.getFileStatus(file1);
+      LocatedBlocks locations = RaidDFSUtil.getBlockLocations(
+        (DistributedFileSystem) fileSys, file1.toUri().getPath(),
+        0, srcStat.getLen());
+
+      DistributedFileSystem dfs = (DistributedFileSystem)fileSys;
+
+      // Corrupt blocks in different stripes. We can fix them.
+      int[] corruptBlockIdxs = new int[]{0, 4, 6};
+      for (int idx: corruptBlockIdxs) {
+        LOG.info("Corrupting block " + locations.get(idx).getBlock());
+        TestRaidDfs.corruptBlock(file1, locations.get(idx).getBlock(),
+                                  NUM_DATANODES, false); // corrupt block
+        long offset = idx * blockSize;
+        try {
+          readAndDiscard(fileSys, file1, offset, blockSize);
+          fail("Expected checksumexception not thrown");
+        } catch (ChecksumException e) {
+          LOG.info("Block at offset " + offset + " got expected exception");
+        }
+      }
+
+      String fileUriPath = file1.toUri().getPath();
+      waitForCorruptBlocks(corruptBlockIdxs.length, dfs, file1);
+
+      // Create RaidShell and fix the file.
+      RaidShell shell = new RaidShell(conf);
+      String[] args = new String[2];
+      args[0] = "-recoverBlocks";
+      args[1] = file1.toUri().getPath();
+      ToolRunner.run(shell, args);
+
+      waitForCorruptBlocks(0, dfs, file1);
+
+      assertTrue(TestRaidDfs.validateFile(dfs, file1, file1Len, crc1));
+
+      // Now corrupt and fix the parity file.
+      FileStatus parityStat = fileSys.getFileStatus(parityFile);
+      long parityCrc = getCRC(fileSys, parityFile);
+      locations = RaidDFSUtil.getBlockLocations(
+        dfs, parityFile.toUri().getPath(), 0, parityStat.getLen());
+      TestRaidDfs.corruptBlock(parityFile, locations.get(0).getBlock(),
+                                NUM_DATANODES, false); // corrupt block
+      try {
+        readAndDiscard(fileSys, parityFile, 0, blockSize);
+        fail("Expected checksumexception not thrown");
+      } catch (ChecksumException e) {
+        LOG.info("Parity Block at offset 0 got expected exception");
+      }
+      waitForCorruptBlocks(1, dfs, parityFile);
+
+      args[1] = parityFile.toUri().getPath();
+      ToolRunner.run(shell, args);
+
+      waitForCorruptBlocks(0, dfs, file1);
+      assertEquals(parityCrc, getCRC(fileSys, parityFile));
+
+    } catch (Exception e) {
+      LOG.info("Test testBlockFix Exception " + e + StringUtils.stringifyException(e));
+      throw e;
+    } finally {
+      myTearDown();
+    }
+    LOG.info("Test testBlockFix completed.");
+  }
+
+  private void waitForCorruptBlocks(
+    int numCorruptBlocks, DistributedFileSystem dfs, Path file)
+    throws Exception {
+    String path = file.toUri().getPath();
+    FileStatus stat = dfs.getFileStatus(file);
+    long start = System.currentTimeMillis();
+    long actual = 0;
+    do {
+      actual = RaidDFSUtil.corruptBlocksInFile(
+          dfs, path, 0, stat.getLen()).size();
+      if (actual == numCorruptBlocks) break;
+      if (System.currentTimeMillis() - start > 120000) break;
+      LOG.info("Waiting for " + numCorruptBlocks + " corrupt blocks in " +
+        path + ", found " + actual);
+      Thread.sleep(1000);
+    } while (true);
+    assertEquals(numCorruptBlocks, actual);
+  }
+
+  private static DistributedFileSystem getDFS(
+        Configuration conf, FileSystem dfs) throws IOException {
+    Configuration clientConf = new Configuration(conf);
+    clientConf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
+    clientConf.setBoolean("fs.hdfs.impl.disable.cache", true);
+    URI dfsUri = dfs.getUri();
+    FileSystem.closeAll();
+    return (DistributedFileSystem) FileSystem.get(dfsUri, clientConf);
+  }
+
+  private void mySetup(int stripeLength, int timeBeforeHar) throws Exception {
+
+    new File(TEST_DIR).mkdirs(); // Make sure data directory exists
+    conf = new Configuration();
+
+    conf.set("raid.config.file", CONFIG_FILE);
+    conf.setBoolean("raid.config.reload", true);
+    conf.setLong("raid.config.reload.interval", RELOAD_INTERVAL);
+
+    // scan all policies once every 5 second
+    conf.setLong("raid.policy.rescan.interval", 5000);
+
+    // make all deletions not go through Trash
+    conf.set("fs.shell.delete.classname", "org.apache.hadoop.hdfs.DFSClient");
+
+    // do not use map-reduce cluster for Raiding
+    conf.setBoolean("fs.raidnode.local", true);
+    conf.set("raid.server.address", "localhost:0");
+    conf.setInt("hdfs.raid.stripeLength", stripeLength);
+    conf.set("hdfs.raid.locations", "/destraid");
+
+    dfs = new MiniDFSCluster(conf, NUM_DATANODES, true, null);
+    dfs.waitActive();
+    fileSys = dfs.getFileSystem();
+    namenode = fileSys.getUri().toString();
+
+    FileSystem.setDefaultUri(conf, namenode);
+    hftp = "hftp://localhost.localdomain:" + dfs.getNameNodePort();
+
+    FileSystem.setDefaultUri(conf, namenode);
+
+    FileWriter fileWriter = new FileWriter(CONFIG_FILE);
+    fileWriter.write("<?xml version=\"1.0\"?>\n");
+    String str = "<configuration> " +
+                   "<srcPath prefix=\"/user/dhruba/raidtest\"> " +
+                     "<policy name = \"RaidTest1\"> " +
+                        "<erasureCode>xor</erasureCode> " +
+                        "<destPath> /destraid</destPath> " +
+                        "<property> " +
+                          "<name>targetReplication</name> " +
+                          "<value>1</value> " +
+                          "<description>after RAIDing, decrease the replication factor of a file to this value." +
+                          "</description> " +
+                        "</property> " +
+                        "<property> " +
+                          "<name>metaReplication</name> " +
+                          "<value>1</value> " +
+                          "<description> replication factor of parity file" +
+                          "</description> " +
+                        "</property> " +
+                        "<property> " +
+                          "<name>modTimePeriod</name> " +
+                          "<value>2000</value> " +
+                          "<description> time (milliseconds) after a file is modified to make it " +
+                                         "a candidate for RAIDing " +
+                          "</description> " +
+                        "</property> ";
+    if (timeBeforeHar >= 0) {
+      str +=
+                        "<property> " +
+                          "<name>time_before_har</name> " +
+                          "<value>" + timeBeforeHar + "</value> " +
+                          "<description> amount of time waited before har'ing parity files" +
+                          "</description> " +
+                        "</property> ";
+    }
+
+    str +=
+                     "</policy>" +
+                   "</srcPath>" +
+                 "</configuration>";
+    fileWriter.write(str);
+    fileWriter.close();
+  }
+
+  private void myTearDown() throws Exception {
+    if (cnode != null) { cnode.stop(); cnode.join(); }
+    if (dfs != null) { dfs.shutdown(); }
+  }
+
+  private long getCRC(FileSystem fs, Path p) throws IOException {
+    CRC32 crc = new CRC32();
+    FSDataInputStream stm = fs.open(p);
+    for (int b = 0; b > 0; b = stm.read()) {
+      crc.update(b);
+    }
+    stm.close();
+    return crc.getValue();
+  }
+
+  private static void readAndDiscard(
+    FileSystem fs, Path p, long offset, long length) throws IOException {
+    FSDataInputStream in = fs.open(p);
+    in.seek(offset);
+    long count = 0;
+    for (int b = 0; b >= 0 && count < length; count++) {
+      b = in.read();
+    }
+  }
+}
+



Mime
View raw message