Skip to content

HDFS-17948. Improve FedBalance DistCp option handling. #8579

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
qijiale76 wants to merge 1 commit into
base: trunk
Choose a base branch
from
Open

HDFS-17948. Improve FedBalance DistCp option handling. #8579

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 69 additions & 0 deletions ...ect/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/rbfbalance/RouterFedBalance.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,12 +51,18 @@
import static org.apache.hadoop.tools.fedbalance.FedBalance.FED_BALANCE_DEFAULT_XML;
import static org.apache.hadoop.tools.fedbalance.FedBalance.FED_BALANCE_SITE_XML;
import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.CLI_OPTIONS;
import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.DISTCP_STRATEGY;
import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.FORCE_CLOSE_OPEN;
import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.LIST_STATUS_THREADS;
import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.MAP;
import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.PRESERVE_TIMES;
import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.SKIP_ACL_PRESERVE;
import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.BANDWIDTH;
import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.DELAY_DURATION;
import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.DIFF_THRESHOLD;
import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.TRASH;
import static org.apache.hadoop.tools.fedbalance.FedBalanceConfigs.PRESERVE_ACL_ENABLED;
import static org.apache.hadoop.tools.fedbalance.FedBalanceConfigs.PRESERVE_ACL_ENABLED_DEFAULT;

/**
* Balance data in router-based federation cluster. From src sub-namespace to
Expand Down Expand Up @@ -92,6 +98,14 @@ private class Builder {
private long delayDuration = TimeUnit.SECONDS.toMillis(1);
/* Specify the threshold of diff entries. */
private int diffThreshold = 0;
/* Whether to preserve ACLs in submitted DistCp jobs. */
private boolean preserveAcl = true;
/* Whether to preserve modification/access times in submitted DistCp jobs. */
private boolean preserveTimes = false;
/* DistCp copy strategy. */
private String distCpStrategy;
/* Number of DistCp listStatus threads. */
private int numListstatusThreads = 0;
/* The source input. This specifies the source path. */
private final String inputSrc;
/* The dst input. This specifies the dst path. */
Expand Down Expand Up @@ -156,6 +170,42 @@ public Builder setDiffThreshold(int value) {
return this;
}

/**
* Specify whether ACLs should be preserved by DistCp.
* @param value true if preserving ACLs.
*/
public Builder setPreserveAcl(boolean value) {
this.preserveAcl = value;
return this;
}

/**
* Specify whether file times should be preserved by DistCp.
* @param value true if preserving file times.
*/
public Builder setPreserveTimes(boolean value) {
this.preserveTimes = value;
return this;
}

/**
* Specify the DistCp copy strategy.
* @param value the DistCp copy strategy.
*/
public Builder setDistCpStrategy(String value) {
this.distCpStrategy = value;
return this;
}

/**
* Specify the DistCp listStatus thread count.
* @param value the DistCp listStatus thread count.
*/
public Builder setNumListstatusThreads(int value) {
this.numListstatusThreads = value;
return this;
}

/**
* Build the balance job.
*/
Expand All @@ -172,6 +222,11 @@ public BalanceJob build() throws IOException {
.setForceCloseOpenFiles(forceCloseOpen).setUseMountReadOnly(true)
.setMapNum(map).setBandwidthLimit(bandwidth).setTrash(trashOpt)
.setDelayDuration(delayDuration).setDiffThreshold(diffThreshold)
.setPreserveAcl(preserveAcl && getConf().getBoolean(
PRESERVE_ACL_ENABLED, PRESERVE_ACL_ENABLED_DEFAULT))
.setPreserveTimes(preserveTimes)
.setDistCpStrategy(distCpStrategy)
.setNumListstatusThreads(numListstatusThreads)
.build();

LOG.info(context.toString());
Expand Down Expand Up @@ -281,6 +336,20 @@ private int submit(CommandLine command, String inputSrc, String inputDst)
builder.setDiffThreshold(Integer.parseInt(
command.getOptionValue(DIFF_THRESHOLD.getOpt())));
}
if (command.hasOption(SKIP_ACL_PRESERVE.getOpt())) {
builder.setPreserveAcl(false);
}
if (command.hasOption(PRESERVE_TIMES.getOpt())) {
builder.setPreserveTimes(true);
}
if (command.hasOption(DISTCP_STRATEGY.getOpt())) {
builder.setDistCpStrategy(command.getOptionValue(
DISTCP_STRATEGY.getOpt()));
}
if (command.hasOption(LIST_STATUS_THREADS.getOpt())) {
builder.setNumListstatusThreads(Integer.parseInt(
command.getOptionValue(LIST_STATUS_THREADS.getOpt())));
}
if (command.hasOption(TRASH.getOpt())) {
String val = command.getOptionValue(TRASH.getOpt());
if (val.equalsIgnoreCase("skip")) {
Expand Down
131 changes: 124 additions & 7 deletions ...-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/DistCpProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.apache.hadoop.fs.permission.AclStatus;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.protocol.AclException;
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.hdfs.protocol.OpenFileEntry;
import org.apache.hadoop.hdfs.protocol.OpenFilesIterator.OpenFilesType;
Expand Down Expand Up @@ -91,9 +92,21 @@ public enum Stage {
private boolean useMountReadOnly;
/* The threshold of diff entries. */
private int diffThreshold;
/* Whether to preserve ACLs in submitted DistCp jobs. */
private boolean preserveAcl;
/* Whether to preserve modification/access times in submitted DistCp jobs. */
private boolean preserveTimes;
/* DistCp copy strategy. */
private String distCpStrategy;
/* Number of DistCp listStatus threads. */
private int numListstatusThreads;

private FsPermission fPerm; // the permission of the src.
private AclStatus acl; // the acl of the src.
/* Cached result of whether ACL is supported on srcFs. Null until checked. */
private Boolean srcAclSupported;
/* Cached result of whether ACL is supported on dstFs. Null until checked. */
private Boolean dstAclSupported;

private JobClient client;
private DistributedFileSystem srcFs; // fs of the src cluster.
Expand Down Expand Up @@ -145,6 +158,10 @@ public DistCpProcedure(String name, String nextProcedure, long delayDuration,
this.forceCloseOpenFiles = context.getForceCloseOpenFiles();
this.useMountReadOnly = context.getUseMountReadOnly();
this.diffThreshold = context.getDiffThreshold();
this.preserveAcl = context.getPreserveAcl();
this.preserveTimes = context.getPreserveTimes();
this.distCpStrategy = context.getDistCpStrategy();
this.numListstatusThreads = context.getNumListstatusThreads();
srcFs = (DistributedFileSystem) context.getSrc().getFileSystem(conf);
dstFs = (DistributedFileSystem) context.getDst().getFileSystem(conf);
}
Expand Down Expand Up @@ -255,7 +272,7 @@ protected void disableWrite(FedBalanceContext fbcontext) throws IOException {
// Save and cancel permission.
FileStatus status = srcFs.getFileStatus(src);
fPerm = status.getPermission();
acl = srcFs.getAclStatus(src);
acl = getAclStatusIfSupported();
srcFs.setPermission(src, FsPermission.createImmutable((short) 0));
updateStage(Stage.FINAL_DISTCP);
}
Expand All @@ -273,15 +290,90 @@ protected void enableWrite() throws IOException {
*/
void restorePermission() throws IOException {
// restore permission.
dstFs.removeAcl(dst);
if (acl != null) {
dstFs.modifyAclEntries(dst, acl.getEntries());
removeAclIfSupported();
if (preserveAcl && acl != null) {
try {
dstFs.modifyAclEntries(dst, acl.getEntries());
} catch (AclException e) {
LOG.info("ACL is not enabled for {}. Skip ACL restore.", dst, e);
}
}
if (fPerm != null) {
dstFs.setPermission(dst, fPerm);
}
}

private AclStatus getAclStatusIfSupported() throws IOException {
if (!shouldPreserveAcl()) {
return null;
}
return srcFs.getAclStatus(src);
}

private void removeAclIfSupported() throws IOException {
if (!preserveAcl || acl == null || !isDstAclSupported()) {
return;
}
try {
dstFs.removeAcl(dst);
} catch (AclException e) {
LOG.info("ACL is not enabled for {}. Skip ACL restore.", dst, e);
}
}

/**
* Whether ACL is supported on the source filesystem. The result is cached
* after the first check since ACL support doesn't change during a job.
*/
private boolean isSrcAclSupported() throws IOException {
if (srcAclSupported == null) {
try {
srcFs.getAclStatus(src);
srcAclSupported = true;
} catch (AclException e) {
LOG.info("ACL is not enabled for {}. Skip ACL preserve.", src, e);
srcAclSupported = false;
}
}
return srcAclSupported;
}

/**
* Whether ACL is supported on the destination filesystem. The destination
* path may not exist before the initial DistCp, so probe the nearest
* existing parent path.
*/
private boolean isDstAclSupported() throws IOException {
if (dstAclSupported == null) {
Path probePath = getAclProbePath(dstFs, dst);
try {
dstFs.getAclStatus(probePath);
dstAclSupported = true;
} catch (AclException e) {
LOG.info("ACL is not enabled for {}. Skip ACL preserve.", probePath,
e);
dstAclSupported = false;
}
}
return dstAclSupported;
}

private Path getAclProbePath(DistributedFileSystem fs, Path path)
throws IOException {
Path probePath = path;
while (probePath != null) {
if (fs.exists(probePath)) {
return probePath;
}
probePath = probePath.getParent();
}
return new Path(Path.SEPARATOR);
}

private boolean shouldPreserveAcl() throws IOException {
return preserveAcl && isSrcAclSupported() && isDstAclSupported();
}

/**
* Close all open files then submit the distcp with -diff.
*/
Expand Down Expand Up @@ -475,8 +567,8 @@ private void pathCheckBeforeInitDistcp() throws IOException {
private String submitDistCpJob(String srcParam, String dstParam,
boolean useSnapshotDiff) throws IOException {
List<String> command = new ArrayList<>();
command.addAll(Arrays
.asList(new String[] {"-async", "-update", "-append", "-pruxgpcab"}));
command.addAll(Arrays.asList(
new String[] {"-async", "-update", "-append", getPreserveOption()}));
if (useSnapshotDiff) {
command.add("-diff");
command.add(LAST_SNAPSHOT_NAME);
Expand All @@ -486,6 +578,14 @@ private String submitDistCpJob(String srcParam, String dstParam,
command.add(mapNum + "");
command.add("-bandwidth");
command.add(bandWidth + "");
if (numListstatusThreads > 0) {
command.add("-numListstatusThreads");
command.add(numListstatusThreads + "");
}
if (distCpStrategy != null && !distCpStrategy.isEmpty()) {
command.add("-strategy");
command.add(distCpStrategy);
}
command.add(srcParam);
command.add(dstParam);

Expand All @@ -505,6 +605,18 @@ private String submitDistCpJob(String srcParam, String dstParam,
}
}

@VisibleForTesting
String getPreserveOption() throws IOException {
StringBuilder preserve = new StringBuilder("-pruxgpcb");
if (shouldPreserveAcl()) {
preserve.append('a');
}
if (preserveTimes) {
preserve.append('t');
}
return preserve.toString();
}

@Override
public void write(DataOutput out) throws IOException {
super.write(out);
Expand Down Expand Up @@ -564,6 +676,11 @@ public void readFields(DataInput in) throws IOException {
bandWidth = context.getBandwidthLimit();
forceCloseOpenFiles = context.getForceCloseOpenFiles();
useMountReadOnly = context.getUseMountReadOnly();
diffThreshold = context.getDiffThreshold();
preserveAcl = context.getPreserveAcl();
preserveTimes = context.getPreserveTimes();
distCpStrategy = context.getDistCpStrategy();
numListstatusThreads = context.getNumListstatusThreads();
this.client = new JobClient(conf);
}

Expand Down Expand Up @@ -666,4 +783,4 @@ public String getFailureInfo() throws IOException {
}
}
}
}
}
Loading
Loading