| /** |
| * 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.namenode; |
| |
| |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertFalse; |
| import static org.junit.Assert.assertTrue; |
| import static org.junit.Assert.fail; |
| import static org.mockito.Mockito.doNothing; |
| import static org.mockito.Mockito.mock; |
| import static org.mockito.Mockito.spy; |
| import static org.mockito.Mockito.when; |
| |
| import java.io.BufferedInputStream; |
| import java.io.File; |
| import java.io.FileInputStream; |
| import java.io.FileOutputStream; |
| import java.io.IOException; |
| import java.io.RandomAccessFile; |
| import java.nio.channels.FileChannel; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.HashMap; |
| import java.util.Map; |
| import java.util.SortedMap; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.fs.permission.FsPermission; |
| import org.apache.hadoop.hdfs.DFSConfigKeys; |
| import org.apache.hadoop.hdfs.DFSTestUtil; |
| import org.apache.hadoop.hdfs.DistributedFileSystem; |
| import org.apache.hadoop.hdfs.HdfsConfiguration; |
| import org.apache.hadoop.hdfs.MiniDFSCluster; |
| import org.apache.hadoop.hdfs.StripedFileTestUtil; |
| import org.apache.hadoop.hdfs.protocol.AddErasureCodingPolicyResponse; |
| import org.apache.hadoop.hdfs.protocol.Block; |
| import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy; |
| import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicyState; |
| import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfo; |
| import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfoContiguous; |
| import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfoStriped; |
| import org.apache.hadoop.hdfs.server.common.HdfsServerConstants; |
| import org.apache.hadoop.hdfs.server.common.Storage.StorageDirectory; |
| import org.apache.hadoop.hdfs.server.namenode.FSEditLogLoader.EditLogValidation; |
| import org.apache.hadoop.hdfs.server.namenode.NNStorage.NameNodeDirType; |
| import org.apache.hadoop.io.IOUtils; |
| import org.apache.hadoop.io.erasurecode.ECSchema; |
| import org.apache.hadoop.test.GenericTestUtils; |
| import org.apache.hadoop.test.GenericTestUtils.LogCapturer; |
| import org.apache.hadoop.test.PathUtils; |
| import org.apache.hadoop.util.FakeTimer; |
| import org.slf4j.event.Level; |
| import org.junit.Test; |
| import org.junit.runner.RunWith; |
| import org.junit.runners.Parameterized; |
| import org.junit.runners.Parameterized.Parameters; |
| |
| import org.apache.hadoop.thirdparty.com.google.common.collect.Maps; |
| import org.apache.hadoop.thirdparty.com.google.common.io.Files; |
| |
| @RunWith(Parameterized.class) |
| public class TestFSEditLogLoader { |
| @Parameters |
| public static Collection<Object[]> data() { |
| Collection<Object[]> params = new ArrayList<Object[]>(); |
| params.add(new Object[]{ Boolean.FALSE }); |
| params.add(new Object[]{ Boolean.TRUE }); |
| return params; |
| } |
| |
| private static boolean useAsyncEditLog; |
| public TestFSEditLogLoader(Boolean async) { |
| useAsyncEditLog = async; |
| } |
| |
| private static Configuration getConf() { |
| Configuration conf = new HdfsConfiguration(); |
| conf.setBoolean(DFSConfigKeys.DFS_NAMENODE_EDITS_ASYNC_LOGGING, |
| useAsyncEditLog); |
| return conf; |
| } |
| |
| static { |
| GenericTestUtils.setLogLevel(FSImage.LOG, Level.TRACE); |
| GenericTestUtils.setLogLevel(FSEditLogLoader.LOG, Level.TRACE); |
| } |
| |
| private static final File TEST_DIR = PathUtils.getTestDir(TestFSEditLogLoader.class); |
| |
| private static final int NUM_DATA_NODES = 0; |
| private static final String FAKE_EDIT_STREAM_NAME = "FAKE_STREAM"; |
| |
| private final ErasureCodingPolicy testECPolicy |
| = StripedFileTestUtil.getDefaultECPolicy(); |
| |
| @Test |
| public void testDisplayRecentEditLogOpCodes() throws IOException { |
| // start a cluster |
| Configuration conf = getConf(); |
| MiniDFSCluster cluster = null; |
| FileSystem fileSys = null; |
| cluster = new MiniDFSCluster.Builder(conf).numDataNodes(NUM_DATA_NODES) |
| .enableManagedDfsDirsRedundancy(false).build(); |
| cluster.waitActive(); |
| fileSys = cluster.getFileSystem(); |
| final FSNamesystem namesystem = cluster.getNamesystem(); |
| |
| FSImage fsimage = namesystem.getFSImage(); |
| for (int i = 0; i < 20; i++) { |
| fileSys.mkdirs(new Path("/tmp/tmp" + i)); |
| } |
| StorageDirectory sd = fsimage.getStorage().dirIterator(NameNodeDirType.EDITS).next(); |
| cluster.shutdown(); |
| |
| File editFile = FSImageTestUtil.findLatestEditsLog(sd).getFile(); |
| assertTrue("Should exist: " + editFile, editFile.exists()); |
| |
| // Corrupt the edits file. |
| long fileLen = editFile.length(); |
| RandomAccessFile rwf = new RandomAccessFile(editFile, "rw"); |
| rwf.seek(fileLen - 40); |
| for (int i = 0; i < 20; i++) { |
| rwf.write(FSEditLogOpCodes.OP_DELETE.getOpCode()); |
| } |
| rwf.close(); |
| |
| StringBuilder bld = new StringBuilder(); |
| bld.append("^Error replaying edit log at offset \\d+. "); |
| bld.append("Expected transaction ID was \\d+\n"); |
| bld.append("Recent opcode offsets: (\\d+\\s*){4}$"); |
| try { |
| cluster = new MiniDFSCluster.Builder(conf).numDataNodes(NUM_DATA_NODES) |
| .enableManagedDfsDirsRedundancy(false).format(false).build(); |
| fail("should not be able to start"); |
| } catch (IOException e) { |
| assertTrue("error message contains opcodes message", |
| e.getMessage().matches(bld.toString())); |
| } |
| } |
| |
| /** |
| * Test that, if the NN restarts with a new minimum replication, |
| * any files created with the old replication count will get |
| * automatically bumped up to the new minimum upon restart. |
| */ |
| @Test |
| public void testReplicationAdjusted() throws Exception { |
| // start a cluster |
| Configuration conf = getConf(); |
| // Replicate and heartbeat fast to shave a few seconds off test |
| conf.setInt(DFSConfigKeys.DFS_NAMENODE_REDUNDANCY_INTERVAL_SECONDS_KEY, 1); |
| conf.setInt(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, 1); |
| |
| MiniDFSCluster cluster = null; |
| try { |
| cluster = new MiniDFSCluster.Builder(conf).numDataNodes(2) |
| .build(); |
| cluster.waitActive(); |
| FileSystem fs = cluster.getFileSystem(); |
| |
| // Create a file with replication count 1 |
| Path p = new Path("/testfile"); |
| DFSTestUtil.createFile(fs, p, 10, /*repl*/ (short)1, 1); |
| DFSTestUtil.waitReplication(fs, p, (short)1); |
| |
| // Shut down and restart cluster with new minimum replication of 2 |
| cluster.shutdown(); |
| cluster = null; |
| |
| conf.setInt(DFSConfigKeys.DFS_NAMENODE_REPLICATION_MIN_KEY, 2); |
| |
| cluster = new MiniDFSCluster.Builder(conf).numDataNodes(2) |
| .format(false).build(); |
| cluster.waitActive(); |
| fs = cluster.getFileSystem(); |
| |
| // The file should get adjusted to replication 2 when |
| // the edit log is replayed. |
| DFSTestUtil.waitReplication(fs, p, (short)2); |
| } finally { |
| if (cluster != null) { |
| cluster.shutdown(); |
| } |
| } |
| } |
| |
| /** |
| * Corrupt the byte at the given offset in the given file, |
| * by subtracting 1 from it. |
| */ |
| private void corruptByteInFile(File file, long offset) |
| throws IOException { |
| RandomAccessFile raf = new RandomAccessFile(file, "rw"); |
| try { |
| raf.seek(offset); |
| int origByte = raf.read(); |
| raf.seek(offset); |
| raf.writeByte(origByte - 1); |
| } finally { |
| IOUtils.closeStream(raf); |
| } |
| } |
| |
| /** |
| * Truncate the given file to the given length |
| */ |
| private void truncateFile(File logFile, long newLength) |
| throws IOException { |
| RandomAccessFile raf = new RandomAccessFile(logFile, "rw"); |
| raf.setLength(newLength); |
| raf.close(); |
| } |
| |
| /** |
| * Return the length of bytes in the given file after subtracting |
| * the trailer of 0xFF (OP_INVALID)s. |
| * This seeks to the end of the file and reads chunks backwards until |
| * it finds a non-0xFF byte. |
| * @throws IOException if the file cannot be read |
| */ |
| private static long getNonTrailerLength(File f) throws IOException { |
| final int chunkSizeToRead = 256*1024; |
| FileInputStream fis = new FileInputStream(f); |
| try { |
| |
| byte buf[] = new byte[chunkSizeToRead]; |
| |
| FileChannel fc = fis.getChannel(); |
| long size = fc.size(); |
| long pos = size - (size % chunkSizeToRead); |
| |
| while (pos >= 0) { |
| fc.position(pos); |
| |
| int readLen = (int) Math.min(size - pos, chunkSizeToRead); |
| IOUtils.readFully(fis, buf, 0, readLen); |
| for (int i = readLen - 1; i >= 0; i--) { |
| if (buf[i] != FSEditLogOpCodes.OP_INVALID.getOpCode()) { |
| return pos + i + 1; // + 1 since we count this byte! |
| } |
| } |
| |
| pos -= chunkSizeToRead; |
| } |
| return 0; |
| } finally { |
| fis.close(); |
| } |
| } |
| |
| @Test |
| public void testStreamLimiter() throws IOException { |
| final File LIMITER_TEST_FILE = new File(TEST_DIR, "limiter.test"); |
| |
| FileOutputStream fos = new FileOutputStream(LIMITER_TEST_FILE); |
| try { |
| fos.write(0x12); |
| fos.write(0x12); |
| fos.write(0x12); |
| } finally { |
| fos.close(); |
| } |
| |
| FileInputStream fin = new FileInputStream(LIMITER_TEST_FILE); |
| BufferedInputStream bin = new BufferedInputStream(fin); |
| FSEditLogLoader.PositionTrackingInputStream tracker = |
| new FSEditLogLoader.PositionTrackingInputStream(bin); |
| try { |
| tracker.setLimit(2); |
| tracker.mark(100); |
| tracker.read(); |
| tracker.read(); |
| try { |
| tracker.read(); |
| fail("expected to get IOException after reading past the limit"); |
| } catch (IOException e) { |
| } |
| tracker.reset(); |
| tracker.mark(100); |
| byte arr[] = new byte[3]; |
| try { |
| tracker.read(arr); |
| fail("expected to get IOException after reading past the limit"); |
| } catch (IOException e) { |
| } |
| tracker.reset(); |
| arr = new byte[2]; |
| tracker.read(arr); |
| } finally { |
| tracker.close(); |
| } |
| } |
| |
| /** |
| * Create an unfinalized edit log for testing purposes |
| * |
| * @param testDir Directory to create the edit log in |
| * @param numTx Number of transactions to add to the new edit log |
| * @param offsetToTxId A map from transaction IDs to offsets in the |
| * edit log file. |
| * @return The new edit log file name. |
| * @throws IOException |
| */ |
| static private File prepareUnfinalizedTestEditLog(File testDir, int numTx, |
| SortedMap<Long, Long> offsetToTxId) throws IOException { |
| File inProgressFile = new File(testDir, NNStorage.getInProgressEditsFileName(1)); |
| FSEditLog fsel = null, spyLog = null; |
| try { |
| fsel = FSImageTestUtil.createStandaloneEditLog(testDir); |
| spyLog = spy(fsel); |
| // Normally, the in-progress edit log would be finalized by |
| // FSEditLog#endCurrentLogSegment. For testing purposes, we |
| // disable that here. |
| doNothing().when(spyLog).endCurrentLogSegment(true); |
| spyLog.openForWrite(NameNodeLayoutVersion.CURRENT_LAYOUT_VERSION); |
| assertTrue("should exist: " + inProgressFile, inProgressFile.exists()); |
| |
| for (int i = 0; i < numTx; i++) { |
| long trueOffset = getNonTrailerLength(inProgressFile); |
| long thisTxId = spyLog.getLastWrittenTxId() + 1; |
| offsetToTxId.put(trueOffset, thisTxId); |
| System.err.println("txid " + thisTxId + " at offset " + trueOffset); |
| spyLog.logDelete("path" + i, i, false); |
| spyLog.logSync(); |
| } |
| } finally { |
| if (spyLog != null) { |
| spyLog.close(); |
| } else if (fsel != null) { |
| fsel.close(); |
| } |
| } |
| return inProgressFile; |
| } |
| |
| @Test |
| public void testValidateEditLogWithCorruptHeader() throws IOException { |
| File testDir = new File(TEST_DIR, "testValidateEditLogWithCorruptHeader"); |
| SortedMap<Long, Long> offsetToTxId = Maps.newTreeMap(); |
| File logFile = prepareUnfinalizedTestEditLog(testDir, 2, offsetToTxId); |
| RandomAccessFile rwf = new RandomAccessFile(logFile, "rw"); |
| try { |
| rwf.seek(0); |
| rwf.writeLong(42); // corrupt header |
| } finally { |
| rwf.close(); |
| } |
| EditLogValidation validation = |
| EditLogFileInputStream.scanEditLog(logFile, Long.MAX_VALUE, true); |
| assertTrue(validation.hasCorruptHeader()); |
| } |
| |
| @Test |
| public void testValidateEditLogWithCorruptBody() throws IOException { |
| File testDir = new File(TEST_DIR, "testValidateEditLogWithCorruptBody"); |
| SortedMap<Long, Long> offsetToTxId = Maps.newTreeMap(); |
| final int NUM_TXNS = 20; |
| File logFile = prepareUnfinalizedTestEditLog(testDir, NUM_TXNS, |
| offsetToTxId); |
| // Back up the uncorrupted log |
| File logFileBak = new File(testDir, logFile.getName() + ".bak"); |
| Files.copy(logFile, logFileBak); |
| EditLogValidation validation = |
| EditLogFileInputStream.scanEditLog(logFile, Long.MAX_VALUE, true); |
| assertTrue(!validation.hasCorruptHeader()); |
| // We expect that there will be an OP_START_LOG_SEGMENT, followed by |
| // NUM_TXNS opcodes, followed by an OP_END_LOG_SEGMENT. |
| assertEquals(NUM_TXNS + 1, validation.getEndTxId()); |
| // Corrupt each edit and verify that validation continues to work |
| for (Map.Entry<Long, Long> entry : offsetToTxId.entrySet()) { |
| long txOffset = entry.getKey(); |
| long txId = entry.getValue(); |
| |
| // Restore backup, corrupt the txn opcode |
| Files.copy(logFileBak, logFile); |
| corruptByteInFile(logFile, txOffset); |
| validation = EditLogFileInputStream.scanEditLog(logFile, |
| Long.MAX_VALUE, true); |
| long expectedEndTxId = (txId == (NUM_TXNS + 1)) ? |
| NUM_TXNS : (NUM_TXNS + 1); |
| assertEquals("Failed when corrupting txn opcode at " + txOffset, |
| expectedEndTxId, validation.getEndTxId()); |
| assertTrue(!validation.hasCorruptHeader()); |
| } |
| |
| // Truncate right before each edit and verify that validation continues |
| // to work |
| for (Map.Entry<Long, Long> entry : offsetToTxId.entrySet()) { |
| long txOffset = entry.getKey(); |
| long txId = entry.getValue(); |
| |
| // Restore backup, corrupt the txn opcode |
| Files.copy(logFileBak, logFile); |
| truncateFile(logFile, txOffset); |
| validation = EditLogFileInputStream.scanEditLog(logFile, |
| Long.MAX_VALUE, true); |
| long expectedEndTxId = (txId == 0) ? |
| HdfsServerConstants.INVALID_TXID : (txId - 1); |
| assertEquals("Failed when corrupting txid " + txId + " txn opcode " + |
| "at " + txOffset, expectedEndTxId, validation.getEndTxId()); |
| assertTrue(!validation.hasCorruptHeader()); |
| } |
| } |
| |
| @Test |
| public void testValidateEmptyEditLog() throws IOException { |
| File testDir = new File(TEST_DIR, "testValidateEmptyEditLog"); |
| SortedMap<Long, Long> offsetToTxId = Maps.newTreeMap(); |
| File logFile = prepareUnfinalizedTestEditLog(testDir, 0, offsetToTxId); |
| // Truncate the file so that there is nothing except the header and |
| // layout flags section. |
| truncateFile(logFile, 8); |
| EditLogValidation validation = |
| EditLogFileInputStream.scanEditLog(logFile, Long.MAX_VALUE, true); |
| assertTrue(!validation.hasCorruptHeader()); |
| assertEquals(HdfsServerConstants.INVALID_TXID, validation.getEndTxId()); |
| } |
| |
| private static final Map<Byte, FSEditLogOpCodes> byteToEnum = |
| new HashMap<Byte, FSEditLogOpCodes>(); |
| static { |
| for(FSEditLogOpCodes opCode : FSEditLogOpCodes.values()) { |
| byteToEnum.put(opCode.getOpCode(), opCode); |
| } |
| } |
| |
| private static FSEditLogOpCodes fromByte(byte opCode) { |
| return byteToEnum.get(opCode); |
| } |
| |
| @Test |
| public void testFSEditLogOpCodes() throws IOException { |
| //try all codes |
| for(FSEditLogOpCodes c : FSEditLogOpCodes.values()) { |
| final byte code = c.getOpCode(); |
| assertEquals("c=" + c + ", code=" + code, |
| c, FSEditLogOpCodes.fromByte(code)); |
| } |
| |
| //try all byte values |
| for(int b = 0; b < (1 << Byte.SIZE); b++) { |
| final byte code = (byte)b; |
| assertEquals("b=" + b + ", code=" + code, |
| fromByte(code), FSEditLogOpCodes.fromByte(code)); |
| } |
| } |
| |
| @Test |
| public void testAddNewStripedBlock() throws IOException{ |
| // start a cluster |
| Configuration conf = new HdfsConfiguration(); |
| MiniDFSCluster cluster = null; |
| try { |
| cluster = new MiniDFSCluster.Builder(conf).numDataNodes(9) |
| .build(); |
| cluster.waitActive(); |
| DistributedFileSystem fs = cluster.getFileSystem(); |
| FSNamesystem fns = cluster.getNamesystem(); |
| fs.enableErasureCodingPolicy(testECPolicy.getName()); |
| |
| String testDir = "/ec"; |
| String testFile = "testfile_001"; |
| String testFilePath = testDir + "/" + testFile; |
| String clientName = "testUser1"; |
| String clientMachine = "testMachine1"; |
| long blkId = 1; |
| long blkNumBytes = 1024; |
| long timestamp = 1426222918; |
| short blockNum = (short) testECPolicy.getNumDataUnits(); |
| short parityNum = (short) testECPolicy.getNumParityUnits(); |
| |
| //set the storage policy of the directory |
| fs.mkdir(new Path(testDir), new FsPermission("755")); |
| fs.getClient().getNamenode().setErasureCodingPolicy( |
| testDir, testECPolicy.getName()); |
| |
| // Create a file with striped block |
| Path p = new Path(testFilePath); |
| DFSTestUtil.createFile(fs, p, 0, (short) 1, 1); |
| |
| fns.enterSafeMode(false); |
| fns.saveNamespace(0, 0); |
| fns.leaveSafeMode(false); |
| |
| // Add a striped block to the file |
| BlockInfoStriped stripedBlk = new BlockInfoStriped( |
| new Block(blkId, blkNumBytes, timestamp), testECPolicy); |
| INodeFile file = (INodeFile)fns.getFSDirectory().getINode(testFilePath); |
| file.toUnderConstruction(clientName, clientMachine); |
| file.addBlock(stripedBlk); |
| fns.getEditLog().logAddBlock(testFilePath, file); |
| TestINodeFile.toCompleteFile(file); |
| |
| //If the block by loaded is the same as above it means that |
| //we have successfully applied the edit log to the fsimage. |
| cluster.restartNameNodes(); |
| cluster.waitActive(); |
| fns = cluster.getNamesystem(); |
| |
| INodeFile inodeLoaded = (INodeFile)fns.getFSDirectory() |
| .getINode(testFilePath); |
| |
| assertTrue(inodeLoaded.isStriped()); |
| |
| BlockInfo[] blks = inodeLoaded.getBlocks(); |
| assertEquals(1, blks.length); |
| assertEquals(blkId, blks[0].getBlockId()); |
| assertEquals(blkNumBytes, blks[0].getNumBytes()); |
| assertEquals(timestamp, blks[0].getGenerationStamp()); |
| assertEquals(blockNum, ((BlockInfoStriped)blks[0]).getDataBlockNum()); |
| assertEquals(parityNum, ((BlockInfoStriped)blks[0]).getParityBlockNum()); |
| |
| cluster.shutdown(); |
| cluster = null; |
| } finally { |
| if (cluster != null) { |
| cluster.shutdown(); |
| } |
| } |
| } |
| |
| @Test |
| public void testUpdateStripedBlocks() throws IOException{ |
| // start a cluster |
| Configuration conf = new HdfsConfiguration(); |
| MiniDFSCluster cluster = null; |
| try { |
| cluster = new MiniDFSCluster.Builder(conf).numDataNodes(9) |
| .build(); |
| cluster.waitActive(); |
| DistributedFileSystem fs = cluster.getFileSystem(); |
| FSNamesystem fns = cluster.getNamesystem(); |
| fs.enableErasureCodingPolicy(testECPolicy.getName()); |
| |
| String testDir = "/ec"; |
| String testFile = "testfile_002"; |
| String testFilePath = testDir + "/" + testFile; |
| String clientName = "testUser2"; |
| String clientMachine = "testMachine2"; |
| long blkId = 1; |
| long blkNumBytes = 1024; |
| long timestamp = 1426222918; |
| short blockNum = (short) testECPolicy.getNumDataUnits(); |
| short parityNum = (short) testECPolicy.getNumParityUnits(); |
| |
| //set the storage policy of the directory |
| fs.mkdir(new Path(testDir), new FsPermission("755")); |
| fs.getClient().getNamenode().setErasureCodingPolicy( |
| testDir, testECPolicy.getName()); |
| |
| //create a file with striped blocks |
| Path p = new Path(testFilePath); |
| DFSTestUtil.createFile(fs, p, 0, (short) 1, 1); |
| BlockInfoStriped stripedBlk = new BlockInfoStriped( |
| new Block(blkId, blkNumBytes, timestamp), testECPolicy); |
| INodeFile file = (INodeFile)fns.getFSDirectory().getINode(testFilePath); |
| file.toUnderConstruction(clientName, clientMachine); |
| file.addBlock(stripedBlk); |
| fns.getEditLog().logAddBlock(testFilePath, file); |
| TestINodeFile.toCompleteFile(file); |
| fns.enterSafeMode(false); |
| fns.saveNamespace(0, 0); |
| fns.leaveSafeMode(false); |
| |
| //update the last block |
| long newBlkNumBytes = 1024*8; |
| long newTimestamp = 1426222918+3600; |
| file.toUnderConstruction(clientName, clientMachine); |
| file.getLastBlock().setNumBytes(newBlkNumBytes); |
| file.getLastBlock().setGenerationStamp(newTimestamp); |
| fns.getEditLog().logUpdateBlocks(testFilePath, file, true); |
| TestINodeFile.toCompleteFile(file); |
| |
| //After the namenode restarts if the block by loaded is the same as above |
| //(new block size and timestamp) it means that we have successfully |
| //applied the edit log to the fsimage. |
| cluster.restartNameNodes(); |
| cluster.waitActive(); |
| fns = cluster.getNamesystem(); |
| |
| INodeFile inodeLoaded = (INodeFile)fns.getFSDirectory() |
| .getINode(testFilePath); |
| |
| assertTrue(inodeLoaded.isStriped()); |
| |
| BlockInfo[] blks = inodeLoaded.getBlocks(); |
| assertEquals(1, blks.length); |
| assertTrue(blks[0].isStriped()); |
| assertEquals(blkId, blks[0].getBlockId()); |
| assertEquals(newBlkNumBytes, blks[0].getNumBytes()); |
| assertEquals(newTimestamp, blks[0].getGenerationStamp()); |
| assertEquals(blockNum, ((BlockInfoStriped)blks[0]).getDataBlockNum()); |
| assertEquals(parityNum, ((BlockInfoStriped)blks[0]).getParityBlockNum()); |
| |
| cluster.shutdown(); |
| cluster = null; |
| } finally { |
| if (cluster != null) { |
| cluster.shutdown(); |
| } |
| } |
| } |
| |
| @Test |
| public void testHasNonEcBlockUsingStripedIDForAddBlock() throws IOException{ |
| // start a cluster |
| Configuration conf = new HdfsConfiguration(); |
| MiniDFSCluster cluster = null; |
| try { |
| cluster = new MiniDFSCluster.Builder(conf).numDataNodes(9) |
| .build(); |
| cluster.waitActive(); |
| DistributedFileSystem fs = cluster.getFileSystem(); |
| FSNamesystem fns = cluster.getNamesystem(); |
| |
| String testDir = "/test_block_manager"; |
| String testFile = "testfile_addblock"; |
| String testFilePath = testDir + "/" + testFile; |
| String clientName = "testUser_addblock"; |
| String clientMachine = "testMachine_addblock"; |
| long blkId = -1; |
| long blkNumBytes = 1024; |
| long timestamp = 1426222918; |
| |
| fs.mkdir(new Path(testDir), new FsPermission("755")); |
| Path p = new Path(testFilePath); |
| |
| //check whether the hasNonEcBlockUsingStripedID is set |
| //after loading a addblock-editlog |
| DFSTestUtil.createFile(fs, p, 0, (short) 1, 1); |
| BlockInfoContiguous cBlk = new BlockInfoContiguous( |
| new Block(blkId, blkNumBytes, timestamp), (short)3); |
| INodeFile file = (INodeFile)fns.getFSDirectory().getINode(testFilePath); |
| file.toUnderConstruction(clientName, clientMachine); |
| file.addBlock(cBlk); |
| fns.getEditLog().logAddBlock(testFilePath, file); |
| TestINodeFile.toCompleteFile(file); |
| cluster.restartNameNodes(); |
| cluster.waitActive(); |
| fns = cluster.getNamesystem(); |
| assertTrue(fns.getBlockManager().hasNonEcBlockUsingStripedID()); |
| |
| cluster.shutdown(); |
| cluster = null; |
| } finally { |
| if (cluster != null) { |
| cluster.shutdown(); |
| } |
| } |
| } |
| |
| @Test |
| public void testHasNonEcBlockUsingStripedIDForUpdateBlocks() |
| throws IOException{ |
| // start a cluster |
| Configuration conf = new HdfsConfiguration(); |
| MiniDFSCluster cluster = null; |
| try { |
| cluster = new MiniDFSCluster.Builder(conf).numDataNodes(9) |
| .build(); |
| cluster.waitActive(); |
| DistributedFileSystem fs = cluster.getFileSystem(); |
| FSNamesystem fns = cluster.getNamesystem(); |
| |
| String testDir = "/test_block_manager"; |
| String testFile = "testfile_002"; |
| String testFilePath = testDir + "/" + testFile; |
| String clientName = "testUser2"; |
| String clientMachine = "testMachine1"; |
| long blkId = 100; |
| long blkNumBytes = 1024; |
| long timestamp = 1426222918; |
| |
| fs.mkdir(new Path(testDir), new FsPermission("755")); |
| Path p = new Path(testFilePath); |
| |
| DFSTestUtil.createFile(fs, p, 0, (short) 1, 1); |
| BlockInfoContiguous cBlk = new BlockInfoContiguous( |
| new Block(blkId, blkNumBytes, timestamp), (short)3); |
| INodeFile file = (INodeFile)fns.getFSDirectory().getINode(testFilePath); |
| file.toUnderConstruction(clientName, clientMachine); |
| file.addBlock(cBlk); |
| TestINodeFile.toCompleteFile(file); |
| |
| long newBlkNumBytes = 1024*8; |
| long newTimestamp = 1426222918+3600; |
| file.toUnderConstruction(clientName, clientMachine); |
| file.getLastBlock().setBlockId(-100); |
| file.getLastBlock().setNumBytes(newBlkNumBytes); |
| file.getLastBlock().setGenerationStamp(newTimestamp); |
| fns.getEditLog().logUpdateBlocks(testFilePath, file, true); |
| TestINodeFile.toCompleteFile(file); |
| cluster.restartNameNodes(); |
| cluster.waitActive(); |
| fns = cluster.getNamesystem(); |
| assertTrue(fns.getBlockManager().hasNonEcBlockUsingStripedID()); |
| |
| cluster.shutdown(); |
| cluster = null; |
| } finally { |
| if (cluster != null) { |
| cluster.shutdown(); |
| } |
| } |
| } |
| |
| @Test |
| public void testErasureCodingPolicyOperations() throws IOException { |
| // start a cluster |
| Configuration conf = new HdfsConfiguration(); |
| final int blockSize = 16 * 1024; |
| conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, blockSize); |
| MiniDFSCluster cluster = null; |
| try { |
| cluster = new MiniDFSCluster.Builder(conf).numDataNodes(9) |
| .build(); |
| cluster.waitActive(); |
| DistributedFileSystem fs = cluster.getFileSystem(); |
| |
| // 1. add new policy |
| ECSchema schema = new ECSchema("rs", 5, 3); |
| int cellSize = 2 * 1024; |
| ErasureCodingPolicy newPolicy = |
| new ErasureCodingPolicy(schema, cellSize, (byte) 0); |
| ErasureCodingPolicy[] policyArray = new ErasureCodingPolicy[]{newPolicy}; |
| AddErasureCodingPolicyResponse[] responses = |
| fs.addErasureCodingPolicies(policyArray); |
| assertEquals(1, responses.length); |
| assertTrue(responses[0].isSucceed()); |
| newPolicy = responses[0].getPolicy(); |
| |
| // Restart NameNode without saving namespace |
| cluster.restartNameNodes(); |
| cluster.waitActive(); |
| |
| // check if new policy is reapplied through edit log |
| ErasureCodingPolicy ecPolicy = |
| ErasureCodingPolicyManager.getInstance().getByID(newPolicy.getId()); |
| assertEquals(ErasureCodingPolicyState.DISABLED, |
| DFSTestUtil.getECPolicyState(ecPolicy)); |
| |
| // 2. enable policy |
| fs.enableErasureCodingPolicy(newPolicy.getName()); |
| cluster.restartNameNodes(); |
| cluster.waitActive(); |
| ecPolicy = |
| ErasureCodingPolicyManager.getInstance().getByID(newPolicy.getId()); |
| assertEquals(ErasureCodingPolicyState.ENABLED, |
| DFSTestUtil.getECPolicyState(ecPolicy)); |
| |
| // create a new file, use the policy |
| final Path dirPath = new Path("/striped"); |
| final Path filePath = new Path(dirPath, "file"); |
| final int fileLength = blockSize * newPolicy.getNumDataUnits(); |
| fs.mkdirs(dirPath); |
| fs.setErasureCodingPolicy(dirPath, newPolicy.getName()); |
| final byte[] bytes = StripedFileTestUtil.generateBytes(fileLength); |
| DFSTestUtil.writeFile(fs, filePath, bytes); |
| |
| // 3. disable policy |
| fs.disableErasureCodingPolicy(newPolicy.getName()); |
| cluster.restartNameNodes(); |
| cluster.waitActive(); |
| ecPolicy = |
| ErasureCodingPolicyManager.getInstance().getByID(newPolicy.getId()); |
| assertEquals(ErasureCodingPolicyState.DISABLED, |
| DFSTestUtil.getECPolicyState(ecPolicy)); |
| // read file |
| DFSTestUtil.readFileAsBytes(fs, filePath); |
| |
| // 4. remove policy |
| fs.removeErasureCodingPolicy(newPolicy.getName()); |
| cluster.restartNameNodes(); |
| cluster.waitActive(); |
| ecPolicy = |
| ErasureCodingPolicyManager.getInstance().getByID(newPolicy.getId()); |
| assertEquals(ErasureCodingPolicyState.REMOVED, |
| DFSTestUtil.getECPolicyState(ecPolicy)); |
| // read file |
| DFSTestUtil.readFileAsBytes(fs, filePath); |
| |
| cluster.shutdown(); |
| cluster = null; |
| } finally { |
| if (cluster != null) { |
| cluster.shutdown(); |
| } |
| } |
| } |
| |
| @Test |
| public void testLoadFSEditLogThrottling() throws Exception { |
| FSNamesystem namesystem = mock(FSNamesystem.class); |
| namesystem.dir = mock(FSDirectory.class); |
| |
| FakeTimer timer = new FakeTimer(); |
| FSEditLogLoader loader = new FSEditLogLoader(namesystem, 0, timer); |
| FSEditLogLoader.LOAD_EDITS_LOG_HELPER.reset(); |
| |
| LogCapturer capture = LogCapturer.captureLogs(FSImage.LOG); |
| loader.loadFSEdits(getFakeEditLogInputStream(1, 10), 1); |
| assertTrue(capture.getOutput().contains("Start loading edits file " + |
| FAKE_EDIT_STREAM_NAME)); |
| assertTrue(capture.getOutput().contains("Loaded 1 edits file(s)")); |
| assertFalse(capture.getOutput().contains("suppressed")); |
| |
| timer.advance(FSEditLogLoader.LOAD_EDIT_LOG_INTERVAL_MS / 2); |
| capture.clearOutput(); |
| loader.loadFSEdits(getFakeEditLogInputStream(11, 20), 11); |
| assertFalse(capture.getOutput().contains("Start loading edits file")); |
| assertFalse(capture.getOutput().contains("edits file(s)")); |
| |
| timer.advance(FSEditLogLoader.LOAD_EDIT_LOG_INTERVAL_MS); |
| capture.clearOutput(); |
| loader.loadFSEdits(getFakeEditLogInputStream(21, 30), 21); |
| assertTrue(capture.getOutput().contains("Start loading edits file " + |
| FAKE_EDIT_STREAM_NAME)); |
| assertTrue(capture.getOutput().contains("suppressed logging 1 times")); |
| assertTrue(capture.getOutput().contains("Loaded 2 edits file(s)")); |
| assertTrue(capture.getOutput().contains("total size 2.0")); |
| } |
| |
| private EditLogInputStream getFakeEditLogInputStream(long startTx, long endTx) |
| throws IOException { |
| EditLogInputStream fakeStream = mock(EditLogInputStream.class); |
| when(fakeStream.getName()).thenReturn(FAKE_EDIT_STREAM_NAME); |
| when(fakeStream.getFirstTxId()).thenReturn(startTx); |
| when(fakeStream.getLastTxId()).thenReturn(endTx); |
| when(fakeStream.length()).thenReturn(1L); |
| return fakeStream; |
| } |
| |
| } |
