| /** |
| * 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 com.google.common.base.Preconditions; |
| import com.google.protobuf.ByteString; |
| import org.apache.commons.io.Charsets; |
| import org.apache.hadoop.HadoopIllegalArgumentException; |
| import org.apache.hadoop.crypto.CipherSuite; |
| import org.apache.hadoop.crypto.CryptoProtocolVersion; |
| import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension; |
| import org.apache.hadoop.fs.CreateFlag; |
| import org.apache.hadoop.fs.FileAlreadyExistsException; |
| import org.apache.hadoop.fs.FileEncryptionInfo; |
| import org.apache.hadoop.fs.InvalidPathException; |
| import org.apache.hadoop.fs.XAttr; |
| import org.apache.hadoop.fs.permission.AclEntry; |
| import org.apache.hadoop.fs.permission.FsAction; |
| import org.apache.hadoop.fs.permission.PermissionStatus; |
| import com.google.protobuf.ByteString; |
| import org.apache.hadoop.fs.InvalidPathException; |
| import org.apache.hadoop.hdfs.DFSConfigKeys; |
| import org.apache.hadoop.hdfs.protocol.Block; |
| import org.apache.hadoop.hdfs.protocol.BlockStoragePolicy; |
| import org.apache.hadoop.hdfs.protocol.ClientProtocol; |
| import org.apache.hadoop.hdfs.protocol.DatanodeInfo; |
| import org.apache.hadoop.hdfs.protocol.EncryptionZone; |
| import org.apache.hadoop.hdfs.protocol.ExtendedBlock; |
| import org.apache.hadoop.hdfs.protocol.HdfsConstants; |
| import org.apache.hadoop.hdfs.protocol.HdfsFileStatus; |
| import org.apache.hadoop.hdfs.protocol.LocatedBlock; |
| import org.apache.hadoop.hdfs.protocol.QuotaExceededException; |
| import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier; |
| import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfoContiguous; |
| import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfoContiguousUnderConstruction; |
| import org.apache.hadoop.hdfs.server.blockmanagement.BlockManager; |
| import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeStorageInfo; |
| import org.apache.hadoop.hdfs.server.common.HdfsServerConstants; |
| import org.apache.hadoop.net.Node; |
| import org.apache.hadoop.net.NodeBase; |
| import org.apache.hadoop.util.ChunkedArrayList; |
| |
| import java.io.FileNotFoundException; |
| import java.io.IOException; |
| import java.nio.ByteBuffer; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.EnumSet; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| |
| import static org.apache.hadoop.hdfs.server.namenode.snapshot.Snapshot.CURRENT_STATE_ID; |
| import static org.apache.hadoop.util.Time.now; |
| |
| class FSDirWriteFileOp { |
| private FSDirWriteFileOp() {} |
| static boolean unprotectedRemoveBlock( |
| FSDirectory fsd, String path, INodesInPath iip, INodeFile fileNode, |
| Block block) throws IOException { |
| // modify file-> block and blocksMap |
| // fileNode should be under construction |
| BlockInfoContiguousUnderConstruction uc = fileNode.removeLastBlock(block); |
| if (uc == null) { |
| return false; |
| } |
| fsd.getBlockManager().removeBlockFromMap(block); |
| |
| if(NameNode.stateChangeLog.isDebugEnabled()) { |
| NameNode.stateChangeLog.debug("DIR* FSDirectory.removeBlock: " |
| +path+" with "+block |
| +" block is removed from the file system"); |
| } |
| |
| // update space consumed |
| fsd.updateCount(iip, 0, -fileNode.getPreferredBlockSize(), |
| uc.getReplication(), true); |
| return true; |
| } |
| |
| /** |
| * Persist the block list for the inode. |
| */ |
| static void persistBlocks( |
| FSDirectory fsd, String path, INodeFile file, boolean logRetryCache) { |
| throw new IllegalStateException("Unimplemented"); |
| } |
| |
| static void persistBlocks( |
| RWTransaction tx, String path, FlatINode inode) { |
| FlatINodeFileFeature f = inode.feature(FlatINodeFileFeature.class); |
| Preconditions.checkArgument(f != null && f.inConstruction()); |
| tx.logUpdateBlocks(path, f); |
| if(NameNode.stateChangeLog.isDebugEnabled()) { |
| NameNode.stateChangeLog.debug( |
| "persistBlocks: " + path + " with " + f.numBlocks() + " " + |
| "blocks is persisted to the file system"); |
| } |
| } |
| |
| static void abandonBlock( |
| FSDirectory fsd, FSPermissionChecker pc, ExtendedBlock b, long fileId, |
| String src, String holder) throws IOException { |
| byte[][] pathComponents = FSDirectory.getPathComponentsForReservedPath(src); |
| src = fsd.resolvePath(pc, src, pathComponents); |
| |
| final INode inode; |
| final INodesInPath iip; |
| if (fileId == HdfsConstants.GRANDFATHER_INODE_ID) { |
| // Older clients may not have given us an inode ID to work with. |
| // In this case, we have to try to resolve the path and hope it |
| // hasn't changed or been deleted since the file was opened for write. |
| iip = fsd.getINodesInPath(src, true); |
| inode = iip.getLastINode(); |
| } else { |
| inode = fsd.getInode(fileId); |
| iip = INodesInPath.fromINode(inode); |
| if (inode != null) { |
| src = iip.getPath(); |
| } |
| } |
| FSNamesystem fsn = fsd.getFSNamesystem(); |
| final INodeFile file = fsn.checkLease(src, holder, inode, fileId); |
| Preconditions.checkState(file.isUnderConstruction()); |
| |
| Block localBlock = ExtendedBlock.getLocalBlock(b); |
| fsd.writeLock(); |
| try { |
| // Remove the block from the pending creates list |
| if (!unprotectedRemoveBlock(fsd, src, iip, file, localBlock)) { |
| return; |
| } |
| } finally { |
| fsd.writeUnlock(); |
| } |
| persistBlocks(fsd, src, file, false); |
| } |
| |
| static void checkBlock(FSNamesystem fsn, ExtendedBlock block) |
| throws IOException { |
| String bpId = fsn.getBlockPoolId(); |
| if (block != null && !bpId.equals(block.getBlockPoolId())) { |
| throw new IOException("Unexpected BlockPoolId " + block.getBlockPoolId() |
| + " - expected " + bpId); |
| } |
| } |
| |
| /** |
| * Part I of getAdditionalBlock(). |
| * Analyze the state of the file under read lock to determine if the client |
| * can add a new block, detect potential retries, lease mismatches, |
| * and minimal replication of the penultimate block. |
| * |
| * Generate target DataNode locations for the new block, |
| * but do not create the new block yet. |
| */ |
| static ValidateAddBlockResult validateAddBlock( |
| FSNamesystem fsn, FSPermissionChecker pc, |
| String src, long fileId, String clientName, |
| ExtendedBlock previous, LocatedBlock[] onRetryBlock) throws IOException { |
| final long blockSize; |
| final int replication; |
| final byte storagePolicyID; |
| String clientMachine; |
| |
| FSDirectory fsd = fsn.getFSDirectory(); |
| try (ROTransaction tx = fsd.newROTransaction().begin()) { |
| FileState fileState = analyzeFileState(tx, fsn, src, fileId, clientName, |
| previous, onRetryBlock); |
| final FlatINode pendingFile = fileState.inode; |
| FlatINodeFileFeature f = pendingFile.feature(FlatINodeFileFeature.class); |
| // Check if the penultimate block is minimally replicated |
| if (!fsn.checkFileProgress(src, f, false)) { |
| throw new NotReplicatedYetException("Not replicated yet: " + src); |
| } |
| |
| if (onRetryBlock[0] != null && onRetryBlock[0].getLocations().length > 0) { |
| // This is a retry. No need to generate new locations. |
| // Use the last block if it has locations. |
| return null; |
| } |
| if (f.numBlocks() >= fsn.maxBlocksPerFile) { |
| throw new IOException("File has reached the limit on maximum number of" |
| + " blocks (" + DFSConfigKeys.DFS_NAMENODE_MAX_BLOCKS_PER_FILE_KEY |
| + "): " + f.numBlocks() + " >= " + fsn.maxBlocksPerFile); |
| } |
| blockSize = f.blockSize(); |
| clientMachine = f.clientMachine(); |
| replication = f.replication(); |
| storagePolicyID = f.storagePolicyId(); |
| return new ValidateAddBlockResult(blockSize, replication, storagePolicyID, |
| clientMachine); |
| } |
| } |
| |
| static LocatedBlock makeLocatedBlock(FSNamesystem fsn, Block blk, |
| DatanodeStorageInfo[] locs, long offset) throws IOException { |
| LocatedBlock lBlk = BlockManager.newLocatedBlock(fsn.getExtendedBlock(blk), |
| locs, offset, false); |
| fsn.getBlockManager().setBlockToken(lBlk, |
| BlockTokenIdentifier.AccessMode.WRITE); |
| return lBlk; |
| } |
| |
| /** |
| * Part II of getAdditionalBlock(). |
| * Should repeat the same analysis of the file state as in Part 1, |
| * but under the write lock. |
| * If the conditions still hold, then allocate a new block with |
| * the new targets, add it to the INode and to the BlocksMap. |
| */ |
| static LocatedBlock storeAllocatedBlock(FSNamesystem fsn, String src, |
| long fileId, String clientName, ExtendedBlock previous, |
| DatanodeStorageInfo[] targets) throws IOException { |
| long offset; |
| // Run the full analysis again, since things could have changed |
| // while chooseTarget() was executing. |
| LocatedBlock[] onRetryBlock = new LocatedBlock[1]; |
| FSDirectory fsd = fsn.getFSDirectory(); |
| BlockManager bm = fsn.getBlockManager(); |
| try (RWTransaction tx = fsd.newRWTransaction().begin()) { |
| FileState fileState = analyzeFileState(tx, fsn, src, fileId, clientName, |
| previous, onRetryBlock); |
| final FlatINode inode = fileState.inode; |
| FlatINodeFileFeature file = inode.feature(FlatINodeFileFeature.class); |
| src = fileState.path; |
| |
| if (onRetryBlock[0] != null) { |
| if (onRetryBlock[0].getLocations().length > 0) { |
| // This is a retry. Just return the last block if having locations. |
| return onRetryBlock[0]; |
| } else { |
| // add new chosen targets to already allocated block and return |
| Block lastBlock = file.lastBlock(); |
| BlockInfoContiguous lastBlockInFile = bm.getStoredBlock(lastBlock); |
| ((BlockInfoContiguousUnderConstruction) lastBlockInFile) |
| .setExpectedLocations(targets); |
| offset = file.fileSize(); |
| return makeLocatedBlock(fsn, lastBlockInFile, targets, offset); |
| } |
| } |
| |
| // commit the last block and complete it if it has minimum replicas |
| FlatINodeFileFeature.Builder newFile = fsn.commitOrCompleteLastBlock( |
| file, ExtendedBlock.getLocalBlock(previous)); |
| |
| // allocate new block, record block locations in INode. |
| Block newBlock = fsn.createNewBlock(); |
| saveAllocatedBlock(fsn, src, inode, newBlock, targets); |
| FlatINode newInode = persistNewBlock(tx, src, inode, newFile, newBlock); |
| offset = newInode.<FlatINodeFileFeature>feature( |
| FlatINodeFileFeature.class).fileSize(); |
| |
| // TODO: Update quota |
| // check quota limits and updated space consumed |
| // fsd.updateCount(inodesInPath, 0, fileINode.getPreferredBlockSize(), |
| // fileINode.getFileReplication(), true); |
| |
| tx.commit(); |
| // Return located block |
| return makeLocatedBlock(fsn, newBlock, targets, offset); |
| } |
| } |
| |
| static DatanodeStorageInfo[] chooseTargetForNewBlock( |
| BlockManager bm, String src, DatanodeInfo[] excludedNodes, String[] |
| favoredNodes, ValidateAddBlockResult r) throws IOException { |
| Node clientNode = bm.getDatanodeManager() |
| .getDatanodeByHost(r.clientMachine); |
| if (clientNode == null) { |
| clientNode = getClientNode(bm, r.clientMachine); |
| } |
| |
| Set<Node> excludedNodesSet = null; |
| if (excludedNodes != null) { |
| excludedNodesSet = new HashSet<>(excludedNodes.length); |
| Collections.addAll(excludedNodesSet, excludedNodes); |
| } |
| List<String> favoredNodesList = (favoredNodes == null) ? null |
| : Arrays.asList(favoredNodes); |
| |
| // choose targets for the new block to be allocated. |
| return bm.chooseTarget4NewBlock(src, r.replication, clientNode, |
| excludedNodesSet, r.blockSize, |
| favoredNodesList, r.storagePolicyID); |
| } |
| |
| /** |
| * Resolve clientmachine address to get a network location path |
| */ |
| static Node getClientNode(BlockManager bm, String clientMachine) { |
| List<String> hosts = new ArrayList<>(1); |
| hosts.add(clientMachine); |
| List<String> rName = bm.getDatanodeManager() |
| .resolveNetworkLocation(hosts); |
| Node clientNode = null; |
| if (rName != null) { |
| // Able to resolve clientMachine mapping. |
| // Create a temp node to findout the rack local nodes |
| clientNode = new NodeBase(rName.get(0) + NodeBase.PATH_SEPARATOR_STR |
| + clientMachine); |
| } |
| return clientNode; |
| } |
| |
| /** |
| * Create a new file or overwrite an existing file<br> |
| * |
| * Once the file is create the client then allocates a new block with the next |
| * call using {@link ClientProtocol#addBlock}. |
| * <p> |
| * For description of parameters and exceptions thrown see |
| * {@link ClientProtocol#create} |
| */ |
| static HdfsFileStatus startFile( |
| FSNamesystem fsn, FSPermissionChecker pc, String src, |
| PermissionStatus permissions, String holder, String clientMachine, |
| EnumSet<CreateFlag> flag, boolean createParent, |
| short replication, long blockSize, |
| EncryptionKeyInfo ezInfo, INode.BlocksMapUpdateInfo toRemoveBlocks, |
| boolean logRetryEntry) |
| throws IOException { |
| assert fsn.hasWriteLock(); |
| |
| boolean create = flag.contains(CreateFlag.CREATE); |
| boolean overwrite = flag.contains(CreateFlag.OVERWRITE); |
| boolean isLazyPersist = flag.contains(CreateFlag.LAZY_PERSIST); |
| |
| CipherSuite suite = null; |
| CryptoProtocolVersion version = null; |
| KeyProviderCryptoExtension.EncryptedKeyVersion edek = null; |
| |
| if (ezInfo != null) { |
| edek = ezInfo.edek; |
| suite = ezInfo.suite; |
| version = ezInfo.protocolVersion; |
| } |
| |
| boolean isRawPath = FSDirectory.isReservedRawName(src); |
| FSDirectory fsd = fsn.getFSDirectory(); |
| final StringMap ugid = fsd.ugid(); |
| |
| try (RWTransaction tx = fsd.newRWTransaction().begin()) { |
| Resolver.Result paths = Resolver.resolve(tx, src); |
| if (paths.invalidPath()) { |
| throw new InvalidPathException(src); |
| } |
| |
| final FlatINodesInPath iip = paths.inodesInPath(); |
| // Verify that the destination does not exist as a directory already. |
| if (paths.ok()) { |
| FlatINode inode = paths.inodesInPath().getLastINode(); |
| if (inode.isDirectory()) { |
| throw new FileAlreadyExistsException(src + |
| " already exists as a directory"); |
| } |
| |
| if (fsd.isPermissionEnabled()) { |
| if (overwrite) { |
| fsd.checkPathAccess(pc, iip, FsAction.WRITE); |
| } |
| } |
| } |
| |
| if (fsd.isPermissionEnabled()) { |
| /* |
| * To overwrite existing file, need to check 'w' permission |
| * of parent (equals to ancestor in this case) |
| */ |
| fsd.checkAncestorAccess(pc, paths, FsAction.WRITE); |
| } |
| |
| if (!createParent && FlatNSUtil.hasNextLevelInPath(paths.src, paths |
| .offset)) { |
| throw new FileNotFoundException(paths.src.substring(0, paths.offset)); |
| } |
| |
| if (paths.notFound() && !create) { |
| throw new FileNotFoundException("Can't overwrite non-existent " + |
| src + " for client " + clientMachine); |
| } |
| |
| // TODO: Handle encryption |
| FileEncryptionInfo feInfo = null; |
| |
| // final EncryptionZone zone = fsd.getEZForPath(iip); |
| // if (zone != null) { |
| // // The path is now within an EZ, but we're missing encryption parameters |
| // if (suite == null || edek == null) { |
| // throw new RetryStartFileException(); |
| // } |
| // // Path is within an EZ and we have provided encryption parameters. |
| // // Make sure that the generated EDEK matches the settings of the EZ. |
| // final String ezKeyName = zone.getKeyName(); |
| // if (!ezKeyName.equals(edek.getEncryptionKeyName())) { |
| // throw new RetryStartFileException(); |
| // } |
| // feInfo = new FileEncryptionInfo(suite, version, |
| // edek.getEncryptedKeyVersion().getMaterial(), |
| // edek.getEncryptedKeyIv(), |
| // ezKeyName, edek.getEncryptionKeyVersionName()); |
| // } |
| |
| if (paths.ok()) { |
| if (overwrite) { |
| // TODO |
| List<Long> toRemoveUCFiles = new ChunkedArrayList<>(); |
| long ret = FSDirDeleteOp.delete(tx, paths, toRemoveBlocks, |
| toRemoveUCFiles, now()); |
| if (ret >= 0) { |
| FSDirDeleteOp.incrDeletedFileCount(ret); |
| fsn.removeLeases(toRemoveUCFiles); |
| } |
| } else { |
| // TODO |
| // If lease soft limit time is expired, recover the lease |
| // fsn.recoverLeaseInternal(FSNamesystem.RecoverLeaseOp.CREATE_FILE, iip, |
| // src, holder, clientMachine, false); |
| throw new FileAlreadyExistsException(src + " for client " + |
| clientMachine + " already exists"); |
| } |
| } |
| fsn.checkFsObjectLimit(); |
| paths = Resolver.resolve(tx, src); |
| Map.Entry<FlatINodesInPath, String> parent = FSDirMkdirOp |
| .createAncestorDirectories(tx, fsd, paths, permissions); |
| long newId = tx.allocateNewInodeId(); |
| FlatINodeFileFeature.Builder fileFeatureBuilder = new FlatINodeFileFeature |
| .Builder() |
| .replication(replication) |
| .blockSize(blockSize) |
| .inConstruction(true) |
| .clientName(holder) |
| .clientMachine(clientMachine); |
| |
| setNewINodeStoragePolicy(fsn.getBlockManager(), fileFeatureBuilder, |
| isLazyPersist); |
| |
| int userId = tx.getStringId(permissions.getUserName()); |
| FlatINode parentINode = parent.getKey().getLastINode(); |
| int groupId = permissions.getGroupName() == null |
| ? parentINode.groupId() |
| : tx.getStringId(permissions.getGroupName()); |
| |
| FlatINodeFileFeature fileFeature = FlatINodeFileFeature.wrap( |
| fileFeatureBuilder.build()); |
| ByteString b = new FlatINode.Builder() |
| .id(newId) |
| .type(FlatINode.Type.FILE) |
| .parentId(parentINode.id()) |
| .mtime(now()) |
| .userId(userId) |
| .groupId(groupId) |
| .permission(permissions.getPermission().toShort()) |
| .addFeature(fileFeature) |
| .build(); |
| |
| FlatINode newNode = FlatINode.wrap(b); |
| // TODO: check .reserved path, quotas and ACL |
| byte[] localName = parent.getValue().getBytes(Charsets.UTF_8); |
| tx.putINode(newId, b); |
| tx.putChild(parentINode.id(), ByteBuffer.wrap(localName), newId); |
| |
| ByteString newParent = new FlatINode.Builder().mergeFrom(parentINode) |
| .mtime(now()).build(); |
| tx.putINode(parentINode.id(), newParent); |
| |
| fsn.leaseManager.addLease(holder, newId); |
| // if (feInfo != null) { |
| // fsd.setFileEncryptionInfo(src, feInfo); |
| // newNode = fsd.getInode(newNode.getId()).asFile(); |
| // } |
| tx.logOpenFile(fsd.ugid(), src, newNode, overwrite, logRetryEntry); |
| if (NameNode.stateChangeLog.isDebugEnabled()) { |
| NameNode.stateChangeLog.debug("DIR* NameSystem.startFile: added " + |
| src + " inode " + newId + " " + holder); |
| } |
| tx.commit(); |
| return FSDirStatAndListingOp.createFileStatus( |
| tx, fsd, newNode, localName, fileFeature.storagePolicyId()); |
| } |
| } |
| |
| static EncryptionKeyInfo getEncryptionKeyInfo(FSNamesystem fsn, |
| FSPermissionChecker pc, String src, |
| CryptoProtocolVersion[] supportedVersions) |
| throws IOException { |
| byte[][] pathComponents = FSDirectory.getPathComponentsForReservedPath(src); |
| FSDirectory fsd = fsn.getFSDirectory(); |
| src = fsd.resolvePath(pc, src, pathComponents); |
| INodesInPath iip = fsd.getINodesInPath4Write(src); |
| // Nothing to do if the path is not within an EZ |
| final EncryptionZone zone = fsd.getEZForPath(iip); |
| if (zone == null) { |
| return null; |
| } |
| CryptoProtocolVersion protocolVersion = fsn.chooseProtocolVersion( |
| zone, supportedVersions); |
| CipherSuite suite = zone.getSuite(); |
| String ezKeyName = zone.getKeyName(); |
| |
| Preconditions.checkNotNull(protocolVersion); |
| Preconditions.checkNotNull(suite); |
| Preconditions.checkArgument(!suite.equals(CipherSuite.UNKNOWN), |
| "Chose an UNKNOWN CipherSuite!"); |
| Preconditions.checkNotNull(ezKeyName); |
| return new EncryptionKeyInfo(protocolVersion, suite, ezKeyName); |
| } |
| |
| static INodeFile addFileForEditLog( |
| FSDirectory fsd, long id, INodesInPath existing, byte[] localName, |
| PermissionStatus permissions, List<AclEntry> aclEntries, |
| List<XAttr> xAttrs, short replication, long modificationTime, long atime, |
| long preferredBlockSize, boolean underConstruction, String clientName, |
| String clientMachine, byte storagePolicyId) { |
| final INodeFile newNode; |
| assert fsd.hasWriteLock(); |
| if (underConstruction) { |
| newNode = newINodeFile(id, permissions, modificationTime, modificationTime, replication, |
| preferredBlockSize, |
| storagePolicyId); |
| newNode.toUnderConstruction(clientName, clientMachine); |
| } else { |
| newNode = newINodeFile(id, permissions, modificationTime, |
| atime, replication, |
| preferredBlockSize, |
| storagePolicyId); |
| } |
| |
| newNode.setLocalName(localName); |
| try { |
| INodesInPath iip = fsd.addINode(existing, newNode); |
| if (iip != null) { |
| if (aclEntries != null) { |
| AclStorage.updateINodeAcl(newNode, aclEntries, CURRENT_STATE_ID); |
| } |
| if (xAttrs != null) { |
| XAttrStorage.updateINodeXAttrs(newNode, xAttrs, CURRENT_STATE_ID); |
| } |
| return newNode; |
| } |
| } catch (IOException e) { |
| if (NameNode.stateChangeLog.isDebugEnabled()) { |
| NameNode.stateChangeLog.debug( |
| "DIR* FSDirectory.unprotectedAddFile: exception when add " |
| + existing.getPath() + " to the file system", e); |
| } |
| } |
| return null; |
| } |
| |
| /** |
| * Add a block to the file. Returns a reference to the added block. |
| */ |
| private static BlockInfoContiguous addBlock( |
| BlockManager bm, String path, FlatINode inode, Block block, |
| DatanodeStorageInfo[] targets) throws IOException { |
| FlatINodeFileFeature f = inode.feature(FlatINodeFileFeature.class); |
| Preconditions.checkState(f.inConstruction()); |
| // associate new last block for the file |
| BlockInfoContiguousUnderConstruction blockInfo = |
| new BlockInfoContiguousUnderConstruction( |
| block, f.replication(), |
| HdfsServerConstants.BlockUCState.UNDER_CONSTRUCTION, |
| targets); |
| |
| bm.addBlockCollection(blockInfo, inode.id()); |
| if(NameNode.stateChangeLog.isDebugEnabled()) { |
| NameNode.stateChangeLog.debug("DIR* FSDirectory.addBlock: " + path |
| + " with " + block + " block is added to the in-memory file system"); |
| } |
| return blockInfo; |
| } |
| |
| private static FileState analyzeFileState( |
| Transaction tx, FSNamesystem fsn, String src, long fileId, String clientName, |
| ExtendedBlock previous, LocatedBlock[] onRetryBlock) |
| throws IOException { |
| assert fsn.hasReadLock(); |
| BlockManager bm = fsn.getBlockManager(); |
| checkBlock(fsn, previous); |
| onRetryBlock[0] = null; |
| |
| Block previousBlock = ExtendedBlock.getLocalBlock(previous); |
| final Resolver.Result paths; |
| if (true || fileId == HdfsConstants.GRANDFATHER_INODE_ID) { |
| // Older clients may not have given us an inode ID to work with. |
| // In this case, we have to try to resolve the path and hope it |
| // hasn't changed or been deleted since the file was opened for write. |
| paths = Resolver.resolve(tx, src); |
| } else { |
| // Newer clients pass the inode ID, so we can just get the inode |
| // directly. |
| paths = Resolver.resolveById(tx, fileId); |
| } |
| if (paths.invalidPath()) { |
| throw new InvalidPathException(src); |
| } else if (paths.notFound()) { |
| throw new FileNotFoundException(src); |
| } |
| FlatINode inode = paths.inodesInPath().getLastINode(); |
| fsn.checkLease(src, clientName, inode); |
| FlatINodeFileFeature pendingFile = inode.feature(FlatINodeFileFeature.class); |
| BlockInfoContiguous lastBlockInFile = pendingFile.lastBlock() == null ? |
| null : bm.getStoredBlock(pendingFile.lastBlock()); |
| if (!Block.matchingIdAndGenStamp(previousBlock, lastBlockInFile)) { |
| // The block that the client claims is the current last block |
| // doesn't match up with what we think is the last block. There are |
| // four possibilities: |
| // 1) This is the first block allocation of an append() pipeline |
| // which started appending exactly at or exceeding the block boundary. |
| // In this case, the client isn't passed the previous block, |
| // so it makes the allocateBlock() call with previous=null. |
| // We can distinguish this since the last block of the file |
| // will be exactly a full block. |
| // 2) This is a retry from a client that missed the response of a |
| // prior getAdditionalBlock() call, perhaps because of a network |
| // timeout, or because of an HA failover. In that case, we know |
| // by the fact that the client is re-issuing the RPC that it |
| // never began to write to the old block. Hence it is safe to |
| // to return the existing block. |
| // 3) This is an entirely bogus request/bug -- we should error out |
| // rather than potentially appending a new block with an empty |
| // one in the middle, etc |
| // 4) This is a retry from a client that timed out while |
| // the prior getAdditionalBlock() is still being processed, |
| // currently working on chooseTarget(). |
| // There are no means to distinguish between the first and |
| // the second attempts in Part I, because the first one hasn't |
| // changed the namesystem state yet. |
| // We run this analysis again in Part II where case 4 is impossible. |
| |
| BlockInfoContiguous penultimateBlock = bm.getStoredBlock( |
| pendingFile.penultimateBlock()); |
| if (previous == null && |
| lastBlockInFile != null && |
| lastBlockInFile.getNumBytes() >= pendingFile.blockSize() && |
| lastBlockInFile.isComplete()) { |
| // Case 1 |
| if (NameNode.stateChangeLog.isDebugEnabled()) { |
| NameNode.stateChangeLog.debug( |
| "BLOCK* NameSystem.allocateBlock: handling block allocation" + |
| " writing to a file with a complete previous block: src=" + |
| src + " lastBlock=" + lastBlockInFile); |
| } |
| } else if (Block.matchingIdAndGenStamp(penultimateBlock, previousBlock)) { |
| if (lastBlockInFile.getNumBytes() != 0) { |
| throw new IOException( |
| "Request looked like a retry to allocate block " + |
| lastBlockInFile + " but it already contains " + |
| lastBlockInFile.getNumBytes() + " bytes"); |
| } |
| |
| // Case 2 |
| // Return the last block. |
| NameNode.stateChangeLog.info("BLOCK* allocateBlock: " + |
| "caught retry for allocation of a new block in " + |
| src + ". Returning previously allocated block " + lastBlockInFile); |
| long offset = pendingFile.fileSize(); |
| BlockInfoContiguousUnderConstruction lastBlockUC = |
| (BlockInfoContiguousUnderConstruction) lastBlockInFile; |
| onRetryBlock[0] = makeLocatedBlock(fsn, lastBlockInFile, |
| lastBlockUC.getExpectedStorageLocations(), offset); |
| return new FileState(inode, src); |
| } else { |
| // Case 3 |
| throw new IOException("Cannot allocate block in " + src + ": " + |
| "passed 'previous' block " + previous + " does not match actual " + |
| "last block in file " + lastBlockInFile); |
| } |
| } |
| return new FileState(inode, src); |
| } |
| |
| static boolean completeFile( |
| FSNamesystem fsn, FSPermissionChecker pc, final String srcArg, |
| String holder, ExtendedBlock last, long fileId) throws IOException { |
| String src = srcArg; |
| if (NameNode.stateChangeLog.isDebugEnabled()) { |
| NameNode.stateChangeLog.debug("DIR* NameSystem.completeFile: " + |
| src + " for " + holder); |
| } |
| checkBlock(fsn, last); |
| byte[][] pathComponents = FSDirectory.getPathComponentsForReservedPath(src); |
| src = fsn.dir.resolvePath(pc, src, pathComponents); |
| boolean success = completeFileInternal(fsn, src, holder, |
| ExtendedBlock.getLocalBlock(last), |
| fileId); |
| if (success) { |
| NameNode.stateChangeLog.info("DIR* completeFile: " + srcArg |
| + " is closed by " + holder); |
| } |
| return success; |
| } |
| |
| private static boolean completeFileInternal( |
| FSNamesystem fsn, String src, String holder, Block last, long fileId) |
| throws IOException { |
| assert fsn.hasWriteLock(); |
| FSDirectory fsd = fsn.getFSDirectory(); |
| try (RWTransaction tx = fsd.newRWTransaction().begin()) { |
| final Resolver.Result paths; |
| if (true || fileId == HdfsConstants.GRANDFATHER_INODE_ID) { |
| // Older clients may not have given us an inode ID to work with. |
| // In this case, we have to try to resolve the path and hope it |
| // hasn't changed or been deleted since the file was opened for write. |
| paths = Resolver.resolve(tx, src); |
| } else { |
| // Newer clients pass the inode ID, so we can just get the inode |
| // directly. |
| paths = Resolver.resolveById(tx, fileId); |
| } |
| if (paths.invalidPath()) { |
| throw new InvalidPathException(src); |
| } else if (paths.notFound()) { |
| throw new FileNotFoundException(src); |
| } |
| FlatINode inode = paths.inodesInPath().getLastINode(); |
| FlatINodeFileFeature file = inode.feature(FlatINodeFileFeature.class); |
| try { |
| fsn.checkLease(src, holder, inode); |
| } catch (LeaseExpiredException lee) { |
| if (file != null && !file.inConstruction()) { |
| // This could be a retry RPC - i.e the client tried to close |
| // the file, but missed the RPC response. Thus, it is trying |
| // again to close the file. If the file still exists and |
| // the client's view of the last block matches the actual |
| // last block, then we'll treat it as a successful close. |
| // See HDFS-3031. |
| final Block realLastBlock = file.lastBlock(); |
| if (Block.matchingIdAndGenStamp(last, realLastBlock)) { |
| NameNode.stateChangeLog.info("DIR* completeFile: request from " |
| + holder + " to complete inode " + fileId + "(" + src + ") " |
| + "which is already closed. But, it appears to be an RPC " |
| + "retry. Returning success"); |
| return true; |
| } |
| } |
| throw lee; |
| } |
| |
| // Check the state of the penultimate block. It should be completed |
| // before attempting to complete the last one. |
| if (!fsn.checkFileProgress(src, file, false)) { |
| return false; |
| } |
| |
| // commit the last block and complete it if it has minimum replicas |
| FlatINodeFileFeature.Builder newFile = |
| fsn.commitOrCompleteLastBlock(file, last); |
| |
| if (!fsn.checkFileProgress(src, file, true)) { |
| return false; |
| } |
| FlatINode.Builder newINode = new FlatINode.Builder().mergeFrom(inode); |
| fsn.finalizeINodeFileUnderConstruction(tx, src, newINode, newFile); |
| tx.commit(); |
| return true; |
| } |
| } |
| |
| private static INodeFile newINodeFile( |
| long id, PermissionStatus permissions, long mtime, long atime, |
| short replication, long preferredBlockSize, byte storagePolicyId) { |
| return new INodeFile(id, null, permissions, mtime, atime, |
| BlockInfoContiguous.EMPTY_ARRAY, replication, preferredBlockSize, |
| storagePolicyId); |
| } |
| |
| /** |
| * Persist the new block (the last block of the given file). |
| */ |
| private static FlatINode persistNewBlock( |
| RWTransaction tx, String path, FlatINode inode, |
| FlatINodeFileFeature.Builder newFile, Block newBlock) { |
| Preconditions.checkArgument(newFile.inConstruction()); |
| newFile.addBlock(newBlock); |
| FlatINodeFileFeature newFeature = FlatINodeFileFeature.wrap(newFile.build()); |
| |
| FlatINode.Builder builder = new FlatINode.Builder() |
| .mergeFrom(inode).replaceFeature(newFeature); |
| ByteString newFileBytes = builder.build(); |
| FlatINode newInode = FlatINode.wrap(newFileBytes); |
| tx.putINode(inode.id(), newFileBytes); |
| tx.logAddBlock(path, newFeature); |
| if (NameNode.stateChangeLog.isDebugEnabled()) { |
| NameNode.stateChangeLog.debug("persistNewBlock: " + path |
| + " with new block " + newBlock + ", current total block count is " |
| + newFeature.numBlocks()); |
| } |
| return newInode; |
| } |
| |
| /** |
| * Save allocated block at the given pending filename |
| * |
| * @param fsn FSNamesystem |
| * @param src path to the file |
| * @param inode the file |
| * @param newBlock newly allocated block to be save |
| * @param targets target datanodes where replicas of the new block is placed |
| * @throws QuotaExceededException If addition of block exceeds space quota |
| */ |
| private static void saveAllocatedBlock( |
| FSNamesystem fsn, String src, FlatINode inode, Block newBlock, |
| DatanodeStorageInfo[] targets) |
| throws IOException { |
| assert fsn.hasWriteLock(); |
| BlockManager bm = fsn.getBlockManager(); |
| BlockInfoContiguous b = addBlock(bm, src, inode, newBlock, targets); |
| NameNode.stateChangeLog.info("BLOCK* allocate " + b + " for " + src); |
| DatanodeStorageInfo.incrementBlocksScheduled(targets); |
| } |
| |
| private static void setNewINodeStoragePolicy( |
| BlockManager bm, FlatINodeFileFeature.Builder file, boolean isLazyPersist) |
| throws IOException { |
| |
| if (isLazyPersist) { |
| BlockStoragePolicy lpPolicy = |
| bm.getStoragePolicy("LAZY_PERSIST"); |
| |
| // Set LAZY_PERSIST storage policy if the flag was passed to |
| // CreateFile. |
| if (lpPolicy == null) { |
| throw new HadoopIllegalArgumentException( |
| "The LAZY_PERSIST storage policy has been disabled " + |
| "by the administrator."); |
| } |
| file.storagePolicyId(lpPolicy.getId()); |
| } else { |
| // TODO: handle effective storage policy id |
| // BlockStoragePolicy effectivePolicy = |
| // bm.getStoragePolicy(parent.getLastINode().storagePolicyId()); |
| // |
| // if (effectivePolicy != null && |
| // effectivePolicy.isCopyOnCreateFile()) { |
| // // Copy effective policy from ancestor directory to current file. |
| // file.storagePolicyId(effectivePolicy.getId()); |
| // } |
| } |
| } |
| |
| private static class FileState { |
| final FlatINode inode; |
| final String path; |
| |
| FileState(FlatINode inode, String fullPath) { |
| this.inode = inode; |
| this.path = fullPath; |
| } |
| } |
| |
| static class ValidateAddBlockResult { |
| final long blockSize; |
| final int replication; |
| final byte storagePolicyID; |
| final String clientMachine; |
| |
| ValidateAddBlockResult( |
| long blockSize, int replication, byte storagePolicyID, |
| String clientMachine) { |
| this.blockSize = blockSize; |
| this.replication = replication; |
| this.storagePolicyID = storagePolicyID; |
| this.clientMachine = clientMachine; |
| } |
| } |
| |
| static class EncryptionKeyInfo { |
| final CryptoProtocolVersion protocolVersion; |
| final CipherSuite suite; |
| final String ezKeyName; |
| KeyProviderCryptoExtension.EncryptedKeyVersion edek; |
| |
| EncryptionKeyInfo( |
| CryptoProtocolVersion protocolVersion, CipherSuite suite, |
| String ezKeyName) { |
| this.protocolVersion = protocolVersion; |
| this.suite = suite; |
| this.ezKeyName = ezKeyName; |
| } |
| } |
| } |
