Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HDFS-7560. ACLs removed by removeDefaultAcl() will be back after NameNode restart/failover. Contributed by Vinayakumar B.

This commit is contained in:
cnauroth 2014-12-22 13:59:10 -08:00
parent a696fbb001
commit 2cf90a2c33
4 changed files with 44 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
View File

@ -627,6 +627,9 @@ Release 2.7.0 - UNRELEASED
HDFS-7557. Fix spacing for a few keys in DFSConfigKeys.java HDFS-7557. Fix spacing for a few keys in DFSConfigKeys.java
(Colin P.McCabe) (Colin P.McCabe)
HDFS-7560. ACLs removed by removeDefaultAcl() will be back after NameNode
restart/failover. (Vinayakumar B via cnauroth)
Release 2.6.1 - UNRELEASED Release 2.6.1 - UNRELEASED
INCOMPATIBLE CHANGES INCOMPATIBLE CHANGES

10
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirAclOp.java
View File

@ -143,7 +143,7 @@ class FSDirAclOp {
try { try {
iip = fsd.getINodesInPath4Write(src); iip = fsd.getINodesInPath4Write(src);
fsd.checkOwner(pc, iip); fsd.checkOwner(pc, iip);
List<AclEntry> newAcl = unprotectedSetAcl(fsd, src, aclSpec); List<AclEntry> newAcl = unprotectedSetAcl(fsd, src, aclSpec, false);
fsd.getEditLog().logSetAcl(src, newAcl); fsd.getEditLog().logSetAcl(src, newAcl);
} finally { } finally {
fsd.writeUnlock(); fsd.writeUnlock();
@ -185,7 +185,7 @@ class FSDirAclOp {
} }
static List<AclEntry> unprotectedSetAcl( static List<AclEntry> unprotectedSetAcl(
FSDirectory fsd, String src, List<AclEntry> aclSpec) FSDirectory fsd, String src, List<AclEntry> aclSpec, boolean fromEdits)
throws IOException { throws IOException {
assert fsd.hasWriteLock(); assert fsd.hasWriteLock();
final INodesInPath iip = fsd.getINodesInPath4Write( final INodesInPath iip = fsd.getINodesInPath4Write(
@ -199,9 +199,11 @@ class FSDirAclOp {
INode inode = FSDirectory.resolveLastINode(iip); INode inode = FSDirectory.resolveLastINode(iip);
int snapshotId = iip.getLatestSnapshotId(); int snapshotId = iip.getLatestSnapshotId();
List<AclEntry> newAcl = aclSpec;
if (!fromEdits) {
List<AclEntry> existingAcl = AclStorage.readINodeLogicalAcl(inode); List<AclEntry> existingAcl = AclStorage.readINodeLogicalAcl(inode);
List<AclEntry> newAcl = AclTransformation.replaceAclEntries(existingAcl, newAcl = AclTransformation.replaceAclEntries(existingAcl, aclSpec);
aclSpec); }
AclStorage.updateINodeAcl(inode, newAcl, snapshotId); AclStorage.updateINodeAcl(inode, newAcl, snapshotId);
return newAcl; return newAcl;
} }

3
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLogLoader.java
View File

@ -823,7 +823,8 @@ public class FSEditLogLoader {
} }
case OP_SET_ACL: { case OP_SET_ACL: {
SetAclOp setAclOp = (SetAclOp) op; SetAclOp setAclOp = (SetAclOp) op;
FSDirAclOp.unprotectedSetAcl(fsDir, setAclOp.src, setAclOp.aclEntries); FSDirAclOp.unprotectedSetAcl(fsDir, setAclOp.src, setAclOp.aclEntries,
true);
break; break;
} }
case OP_SET_XATTR: { case OP_SET_XATTR: {

39
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/FSAclBaseTest.java
View File

@ -426,7 +426,7 @@ public abstract class FSAclBaseTest {
} }
@Test @Test
public void testRemoveDefaultAcl() throws IOException { public void testRemoveDefaultAcl() throws Exception {
FileSystem.mkdirs(fs, path, FsPermission.createImmutable((short)0750)); FileSystem.mkdirs(fs, path, FsPermission.createImmutable((short)0750));
List<AclEntry> aclSpec = Lists.newArrayList( List<AclEntry> aclSpec = Lists.newArrayList(
aclEntry(ACCESS, USER, ALL), aclEntry(ACCESS, USER, ALL),
@ -443,10 +443,15 @@ public abstract class FSAclBaseTest {
aclEntry(ACCESS, GROUP, READ_EXECUTE) }, returned); aclEntry(ACCESS, GROUP, READ_EXECUTE) }, returned);
assertPermission((short)010770); assertPermission((short)010770);
assertAclFeature(true); assertAclFeature(true);
// restart of the cluster
restartCluster();
s = fs.getAclStatus(path);
AclEntry[] afterRestart = s.getEntries().toArray(new AclEntry[0]);
assertArrayEquals(returned, afterRestart);
} }
@Test @Test
public void testRemoveDefaultAclOnlyAccess() throws IOException { public void testRemoveDefaultAclOnlyAccess() throws Exception {
fs.create(path).close(); fs.create(path).close();
fs.setPermission(path, FsPermission.createImmutable((short)0640)); fs.setPermission(path, FsPermission.createImmutable((short)0640));
List<AclEntry> aclSpec = Lists.newArrayList( List<AclEntry> aclSpec = Lists.newArrayList(
@ -463,10 +468,15 @@ public abstract class FSAclBaseTest {
aclEntry(ACCESS, GROUP, READ_EXECUTE) }, returned); aclEntry(ACCESS, GROUP, READ_EXECUTE) }, returned);
assertPermission((short)010770); assertPermission((short)010770);
assertAclFeature(true); assertAclFeature(true);
// restart of the cluster
restartCluster();
s = fs.getAclStatus(path);
AclEntry[] afterRestart = s.getEntries().toArray(new AclEntry[0]);
assertArrayEquals(returned, afterRestart);
} }
@Test @Test
public void testRemoveDefaultAclOnlyDefault() throws IOException { public void testRemoveDefaultAclOnlyDefault() throws Exception {
FileSystem.mkdirs(fs, path, FsPermission.createImmutable((short)0750)); FileSystem.mkdirs(fs, path, FsPermission.createImmutable((short)0750));
List<AclEntry> aclSpec = Lists.newArrayList( List<AclEntry> aclSpec = Lists.newArrayList(
aclEntry(DEFAULT, USER, "foo", ALL)); aclEntry(DEFAULT, USER, "foo", ALL));
@ -477,10 +487,15 @@ public abstract class FSAclBaseTest {
assertArrayEquals(new AclEntry[] { }, returned); assertArrayEquals(new AclEntry[] { }, returned);
assertPermission((short)0750); assertPermission((short)0750);
assertAclFeature(false); assertAclFeature(false);
// restart of the cluster
restartCluster();
s = fs.getAclStatus(path);
AclEntry[] afterRestart = s.getEntries().toArray(new AclEntry[0]);
assertArrayEquals(returned, afterRestart);
} }
@Test @Test
public void testRemoveDefaultAclMinimal() throws IOException { public void testRemoveDefaultAclMinimal() throws Exception {
FileSystem.mkdirs(fs, path, FsPermission.createImmutable((short)0750)); FileSystem.mkdirs(fs, path, FsPermission.createImmutable((short)0750));
fs.removeDefaultAcl(path); fs.removeDefaultAcl(path);
AclStatus s = fs.getAclStatus(path); AclStatus s = fs.getAclStatus(path);
@ -488,10 +503,15 @@ public abstract class FSAclBaseTest {
assertArrayEquals(new AclEntry[] { }, returned); assertArrayEquals(new AclEntry[] { }, returned);
assertPermission((short)0750); assertPermission((short)0750);
assertAclFeature(false); assertAclFeature(false);
// restart of the cluster
restartCluster();
s = fs.getAclStatus(path);
AclEntry[] afterRestart = s.getEntries().toArray(new AclEntry[0]);
assertArrayEquals(returned, afterRestart);
} }
@Test @Test
public void testRemoveDefaultAclStickyBit() throws IOException { public void testRemoveDefaultAclStickyBit() throws Exception {
FileSystem.mkdirs(fs, path, FsPermission.createImmutable((short)01750)); FileSystem.mkdirs(fs, path, FsPermission.createImmutable((short)01750));
List<AclEntry> aclSpec = Lists.newArrayList( List<AclEntry> aclSpec = Lists.newArrayList(
aclEntry(ACCESS, USER, ALL), aclEntry(ACCESS, USER, ALL),
@ -508,6 +528,11 @@ public abstract class FSAclBaseTest {
aclEntry(ACCESS, GROUP, READ_EXECUTE) }, returned); aclEntry(ACCESS, GROUP, READ_EXECUTE) }, returned);
assertPermission((short)011770); assertPermission((short)011770);
assertAclFeature(true); assertAclFeature(true);
// restart of the cluster
restartCluster();
s = fs.getAclStatus(path);
AclEntry[] afterRestart = s.getEntries().toArray(new AclEntry[0]);
assertArrayEquals(returned, afterRestart);
} }
@Test(expected=FileNotFoundException.class) @Test(expected=FileNotFoundException.class)
@ -1137,9 +1162,7 @@ public abstract class FSAclBaseTest {
assertFilePermissionDenied(fsAsDiana, DIANA, bruceFile); assertFilePermissionDenied(fsAsDiana, DIANA, bruceFile);
try { try {
conf.setBoolean(DFSConfigKeys.DFS_PERMISSIONS_ENABLED_KEY, false); conf.setBoolean(DFSConfigKeys.DFS_PERMISSIONS_ENABLED_KEY, false);
destroyFileSystems();
restartCluster(); restartCluster();
initFileSystems();
assertFilePermissionGranted(fsAsDiana, DIANA, bruceFile); assertFilePermissionGranted(fsAsDiana, DIANA, bruceFile);
} finally { } finally {
conf.setBoolean(DFSConfigKeys.DFS_PERMISSIONS_ENABLED_KEY, true); conf.setBoolean(DFSConfigKeys.DFS_PERMISSIONS_ENABLED_KEY, true);
@ -1404,10 +1427,12 @@ public abstract class FSAclBaseTest {
* @throws Exception if restart fails * @throws Exception if restart fails
*/ */
private void restartCluster() throws Exception { private void restartCluster() throws Exception {
destroyFileSystems();
shutdown(); shutdown();
cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).format(false) cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).format(false)
.build(); .build();
cluster.waitActive(); cluster.waitActive();
initFileSystems();
} }
/** /**