import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
@Override
public byte[] retrievePassword(
    DelegationTokenIdentifier identifier) throws InvalidToken {
  try {
    // this check introduces inconsistency in the authentication to a
    // HA standby NN.  non-token auths are allowed into the namespace which
    // decides whether to throw a StandbyException.  tokens are a bit
    // different in that a standby may be behind and thus not yet know
    // of all tokens issued by the active NN.  the following check does
    // not allow ANY token auth, however it should allow known tokens in
    namesystem.checkOperation(OperationCategory.READ);
  } catch (StandbyException se) {
    // FIXME: this is a hack to get around changing method signatures by
    // tunneling a non-InvalidToken exception as the cause which the
    // RPC server will unwrap before returning to the client
    InvalidToken wrappedStandby = new InvalidToken("StandbyException");
    wrappedStandby.initCause(se);
    throw wrappedStandby;
  }
  return super.retrievePassword(identifier);
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
@Override
public byte[] retriableRetrievePassword(DelegationTokenIdentifier identifier)
    throws InvalidToken, StandbyException, RetriableException, IOException {
  namesystem.checkOperation(OperationCategory.READ);
  try {
    return super.retrievePassword(identifier);
  } catch (InvalidToken it) {
    if (namesystem.inTransitionToActive()) {
      // if the namesystem is currently in the middle of transition to 
      // active state, let client retry since the corresponding editlog may 
      // have not been applied yet
      throw new RetriableException(it);
    } else {
      throw it;
    }
  }
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
/**
 * Dump all metadata into specified file
 */
void metaSave(String filename) throws IOException {
  checkSuperuserPrivilege();
  checkOperation(OperationCategory.UNCHECKED);
  writeLock();
  try {
    checkOperation(OperationCategory.UNCHECKED);
    File file = new File(System.getProperty("hadoop.log.dir"), filename);
    PrintWriter out = new PrintWriter(new BufferedWriter(
        new OutputStreamWriter(new FileOutputStream(file), Charsets.UTF_8)));
    metaSave(out);
    out.flush();
    out.close();
  } finally {
    writeUnlock();
  }
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
/**
 * Set permissions for an existing file.
 * @throws IOException
 */
void setPermission(String src, FsPermission permission) throws IOException {
  HdfsFileStatus auditStat;
  checkOperation(OperationCategory.WRITE);
  writeLock();
  try {
    checkOperation(OperationCategory.WRITE);
    checkNameNodeSafeMode("Cannot set permission for " + src);
    auditStat = FSDirAttrOp.setPermission(dir, src, permission);
  } catch (AccessControlException e) {
    logAuditEvent(false, "setPermission", src);
    throw e;
  } finally {
    writeUnlock();
  }
  getEditLog().logSync();
  logAuditEvent(true, "setPermission", src, null, auditStat);
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
/**
 * Set owner for an existing file.
 * @throws IOException
 */
void setOwner(String src, String username, String group)
    throws IOException {
  HdfsFileStatus auditStat;
  checkOperation(OperationCategory.WRITE);
  writeLock();
  try {
    checkOperation(OperationCategory.WRITE);
    checkNameNodeSafeMode("Cannot set owner for " + src);
    auditStat = FSDirAttrOp.setOwner(dir, src, username, group);
  } catch (AccessControlException e) {
    logAuditEvent(false, "setOwner", src);
    throw e;
  } finally {
    writeUnlock();
  }
  getEditLog().logSync();
  logAuditEvent(true, "setOwner", src, null, auditStat);
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
/**
 * Moves all the blocks from {@code srcs} and appends them to {@code target}
 * To avoid rollbacks we will verify validity of ALL of the args
 * before we start actual move.
 * 
 * This does not support ".inodes" relative path
 * @param target target to concat into
 * @param srcs file that will be concatenated
 * @throws IOException on error
 */
void concat(String target, String [] srcs, boolean logRetryCache)
    throws IOException {
  checkOperation(OperationCategory.WRITE);
  waitForLoadingFSImage();
  HdfsFileStatus stat = null;
  boolean success = false;
  writeLock();
  try {
    checkOperation(OperationCategory.WRITE);
    checkNameNodeSafeMode("Cannot concat " + target);
    stat = FSDirConcatOp.concat(dir, target, srcs, logRetryCache);
    success = true;
  } finally {
    writeUnlock();
    if (success) {
      getEditLog().logSync();
    }
    logAuditEvent(success, "concat", Arrays.toString(srcs), target, stat);
  }
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
/**
 * stores the modification and access time for this inode. 
 * The access time is precise up to an hour. The transaction, if needed, is
 * written to the edits log but is not flushed.
 */
void setTimes(String src, long mtime, long atime) throws IOException {
  HdfsFileStatus auditStat;
  checkOperation(OperationCategory.WRITE);
  writeLock();
  try {
    checkOperation(OperationCategory.WRITE);
    checkNameNodeSafeMode("Cannot set times " + src);
    auditStat = FSDirAttrOp.setTimes(dir, src, mtime, atime);
  } catch (AccessControlException e) {
    logAuditEvent(false, "setTimes", src);
    throw e;
  } finally {
    writeUnlock();
  }
  getEditLog().logSync();
  logAuditEvent(true, "setTimes", src, null, auditStat);
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
/**
 * Create a symbolic link.
 */
@SuppressWarnings("deprecation")
void createSymlink(String target, String link,
    PermissionStatus dirPerms, boolean createParent, boolean logRetryCache)
    throws IOException {
  if (!FileSystem.areSymlinksEnabled()) {
    throw new UnsupportedOperationException("Symlinks not supported");
  }
  HdfsFileStatus auditStat = null;
  checkOperation(OperationCategory.WRITE);
  writeLock();
  try {
    checkOperation(OperationCategory.WRITE);
    checkNameNodeSafeMode("Cannot create symlink " + link);
    auditStat = FSDirSymlinkOp.createSymlinkInt(this, target, link, dirPerms,
                                                createParent, logRetryCache);
  } catch (AccessControlException e) {
    logAuditEvent(false, "createSymlink", link, target, null);
    throw e;
  } finally {
    writeUnlock();
  }
  getEditLog().logSync();
  logAuditEvent(true, "createSymlink", link, target, auditStat);
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
/**
 * Set replication for an existing file.
 * 
 * The NameNode sets new replication and schedules either replication of 
 * under-replicated data blocks or removal of the excessive block copies 
 * if the blocks are over-replicated.
 * 
 * @see ClientProtocol#setReplication(String, short)
 * @param src file name
 * @param replication new replication
 * @return true if successful; 
 *         false if file does not exist or is a directory
 */
boolean setReplication(final String src, final short replication)
    throws IOException {
  boolean success = false;
  waitForLoadingFSImage();
  checkOperation(OperationCategory.WRITE);
  writeLock();
  try {
    checkOperation(OperationCategory.WRITE);
    checkNameNodeSafeMode("Cannot set replication for " + src);
    success = FSDirAttrOp.setReplication(dir, blockManager, src, replication);
  } catch (AccessControlException e) {
    logAuditEvent(false, "setReplication", src);
    throw e;
  } finally {
    writeUnlock();
  }
  if (success) {
    getEditLog().logSync();
    logAuditEvent(true, "setReplication", src);
  }
  return success;
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
/**
 * Set the storage policy for a file or a directory.
 *
 * @param src file/directory path
 * @param policyName storage policy name
 */
void setStoragePolicy(String src, String policyName) throws IOException {
  HdfsFileStatus auditStat;
  waitForLoadingFSImage();
  checkOperation(OperationCategory.WRITE);
  writeLock();
  try {
    checkOperation(OperationCategory.WRITE);
    checkNameNodeSafeMode("Cannot set storage policy for " + src);
    auditStat = FSDirAttrOp.setStoragePolicy(
        dir, blockManager, src, policyName);
  } catch (AccessControlException e) {
    logAuditEvent(false, "setStoragePolicy", src);
    throw e;
  } finally {
    writeUnlock();
  }
  getEditLog().logSync();
  logAuditEvent(true, "setStoragePolicy", src, null, auditStat);
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
/** 
 * Change the indicated filename. 
 * @deprecated Use {@link #renameTo(String, String, boolean,
 * Options.Rename...)} instead.
 */
@Deprecated
boolean renameTo(String src, String dst, boolean logRetryCache)
    throws IOException {
  waitForLoadingFSImage();
  checkOperation(OperationCategory.WRITE);
  FSDirRenameOp.RenameOldResult ret = null;
  writeLock();
  try {
    checkOperation(OperationCategory.WRITE);
    checkNameNodeSafeMode("Cannot rename " + src);
    ret = FSDirRenameOp.renameToInt(dir, src, dst, logRetryCache);
  } catch (AccessControlException e)  {
    logAuditEvent(false, "rename", src, dst, null);
    throw e;
  } finally {
    writeUnlock();
  }
  boolean success = ret != null && ret.success;
  if (success) {
    getEditLog().logSync();
  }
  logAuditEvent(success, "rename", src, dst,
      ret == null ? null : ret.auditStat);
  return success;
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
/**
 * Get the file info for a specific file.
 *
 * @param src The string representation of the path to the file
 * @param resolveLink whether to throw UnresolvedLinkException
 *        if src refers to a symlink
 *
 * @throws AccessControlException if access is denied
 * @throws UnresolvedLinkException if a symlink is encountered.
 *
 * @return object containing information regarding the file
 *         or null if file not found
 * @throws StandbyException
 */
HdfsFileStatus getFileInfo(final String src, boolean resolveLink)
  throws IOException {
  checkOperation(OperationCategory.READ);
  HdfsFileStatus stat = null;
  readLock();
  try {
    checkOperation(OperationCategory.READ);
    stat = FSDirStatAndListingOp.getFileInfo(dir, src, resolveLink);
  } catch (AccessControlException e) {
    logAuditEvent(false, "getfileinfo", src);
    throw e;
  } finally {
    readUnlock();
  }
  logAuditEvent(true, "getfileinfo", src);
  return stat;
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
/**
 * Create all the necessary directories
 */
boolean mkdirs(String src, PermissionStatus permissions,
    boolean createParent) throws IOException {
  HdfsFileStatus auditStat = null;
  checkOperation(OperationCategory.WRITE);
  writeLock();
  try {
    checkOperation(OperationCategory.WRITE);
    checkNameNodeSafeMode("Cannot create directory " + src);
    auditStat = FSDirMkdirOp.mkdirs(this, src, permissions, createParent);
  } catch (AccessControlException e) {
    logAuditEvent(false, "mkdirs", src);
    throw e;
  } finally {
    writeUnlock();
  }
  getEditLog().logSync();
  logAuditEvent(true, "mkdirs", src, null, auditStat);
  return true;
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
/**
 * Set the namespace quota and storage space quota for a directory.
 * See {@link ClientProtocol#setQuota(String, long, long, StorageType)} for the
 * contract.
 * 
 * Note: This does not support ".inodes" relative path.
 */
void setQuota(String src, long nsQuota, long ssQuota, StorageType type)
    throws IOException {
  checkOperation(OperationCategory.WRITE);
  writeLock();
  boolean success = false;
  try {
    checkOperation(OperationCategory.WRITE);
    checkNameNodeSafeMode("Cannot set quota on " + src);
    FSDirAttrOp.setQuota(dir, src, nsQuota, ssQuota, type);
    success = true;
  } finally {
    writeUnlock();
    if (success) {
      getEditLog().logSync();
    }
    logAuditEvent(success, "setQuota", src);
  }
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
/**
 * Get a partial listing of the indicated directory
 *
 * @param src the directory name
 * @param startAfter the name to start after
 * @param needLocation if blockLocations need to be returned
 * @return a partial listing starting after startAfter
 * 
 * @throws AccessControlException if access is denied
 * @throws UnresolvedLinkException if symbolic link is encountered
 * @throws IOException if other I/O error occurred
 */
DirectoryListing getListing(String src, byte[] startAfter,
    boolean needLocation) 
    throws IOException {
  checkOperation(OperationCategory.READ);
  DirectoryListing dl = null;
  readLock();
  try {
    checkOperation(NameNode.OperationCategory.READ);
    dl = FSDirStatAndListingOp.getListingInt(dir, src, startAfter,
        needLocation);
  } catch (AccessControlException e) {
    logAuditEvent(false, "listStatus", src);
    throw e;
  } finally {
    readUnlock();
  }
  logAuditEvent(true, "listStatus", src);
  return dl;
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
DatanodeInfo[] datanodeReport(final DatanodeReportType type
    ) throws AccessControlException, StandbyException {
  checkSuperuserPrivilege();
  checkOperation(OperationCategory.UNCHECKED);
  readLock();
  try {
    checkOperation(OperationCategory.UNCHECKED);
    final DatanodeManager dm = getBlockManager().getDatanodeManager();      
    final List<DatanodeDescriptor> results = dm.getDatanodeListForReport(type);

    DatanodeInfo[] arr = new DatanodeInfo[results.size()];
    for (int i=0; i<arr.length; i++) {
      arr[i] = new DatanodeInfo(results.get(i));
    }
    return arr;
  } finally {
    readUnlock();
  }
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
DatanodeStorageReport[] getDatanodeStorageReport(final DatanodeReportType type
    ) throws AccessControlException, StandbyException {
  checkSuperuserPrivilege();
  checkOperation(OperationCategory.UNCHECKED);
  readLock();
  try {
    checkOperation(OperationCategory.UNCHECKED);
    final DatanodeManager dm = getBlockManager().getDatanodeManager();      
    final List<DatanodeDescriptor> datanodes = dm.getDatanodeListForReport(type);

    DatanodeStorageReport[] reports = new DatanodeStorageReport[datanodes.size()];
    for (int i = 0; i < reports.length; i++) {
      final DatanodeDescriptor d = datanodes.get(i);
      reports[i] = new DatanodeStorageReport(new DatanodeInfo(d),
          d.getStorageReports());
    }
    return reports;
  } finally {
    readUnlock();
  }
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
/**
 * Save namespace image.
 * This will save current namespace into fsimage file and empty edits file.
 * Requires superuser privilege and safe mode.
 * 
 * @throws AccessControlException if superuser privilege is violated.
 * @throws IOException if 
 */
void saveNamespace() throws AccessControlException, IOException {
  checkOperation(OperationCategory.UNCHECKED);
  checkSuperuserPrivilege();

  cpLock();  // Block if a checkpointing is in progress on standby.
  readLock();
  try {
    checkOperation(OperationCategory.UNCHECKED);

    if (!isInSafeMode()) {
      throw new IOException("Safe mode should be turned ON "
          + "in order to create namespace image.");
    }
    getFSImage().saveNamespace(this);
  } finally {
    readUnlock();
    cpUnlock();
  }
  LOG.info("New namespace image has been created");
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
/**
 * Enables/Disables/Checks restoring failed storage replicas if the storage becomes available again.
 * Requires superuser privilege.
 * 
 * @throws AccessControlException if superuser privilege is violated.
 */
boolean restoreFailedStorage(String arg) throws AccessControlException,
    StandbyException {
  checkSuperuserPrivilege();
  checkOperation(OperationCategory.UNCHECKED);
  cpLock();  // Block if a checkpointing is in progress on standby.
  writeLock();
  try {
    checkOperation(OperationCategory.UNCHECKED);
    
    // if it is disabled - enable it and vice versa.
    if(arg.equals("check"))
      return getFSImage().getStorage().getRestoreFailedStorage();
    
    boolean val = arg.equals("true");  // false if not
    getFSImage().getStorage().setRestoreFailedStorage(val);
    
    return val;
  } finally {
    writeUnlock();
    cpUnlock();
  }
}
 

import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory; //导入依赖的package包/类
NamenodeCommand startCheckpoint(NamenodeRegistration backupNode,
    NamenodeRegistration activeNamenode) throws IOException {
  checkOperation(OperationCategory.CHECKPOINT);
  writeLock();
  try {
    checkOperation(OperationCategory.CHECKPOINT);
    checkNameNodeSafeMode("Checkpoint not started");
    
    LOG.info("Start checkpoint for " + backupNode.getAddress());
    NamenodeCommand cmd = getFSImage().startCheckpoint(backupNode,
        activeNamenode);
    getEditLog().logSync();
    return cmd;
  } finally {
    writeUnlock();
  }
}