big-data/hadoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

HDFS-14833. RBF: Router Update Doesn't Sync Quota. Contributed by Ayush Saxena.

Browse Source
This commit is contained in:
Ayush Saxena 2019-09-21 09:53:13 +05:30
parent d7d6ec8fae
commit aa938662f9
2 changed files with 74 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterAdminServer.java
View File

@ -269,16 +269,21 @@ public AddMountTableEntryResponse addMountTableEntry(
@Override
public UpdateMountTableEntryResponse updateMountTableEntry(
UpdateMountTableEntryRequest request) throws IOException {
MountTable updateEntry = request.getEntry();
MountTable oldEntry = null;
if (this.router.getSubclusterResolver() instanceof MountTableResolver) {
MountTableResolver mResolver =
(MountTableResolver) this.router.getSubclusterResolver();
oldEntry = mResolver.getMountPoint(updateEntry.getSourcePath());
}
UpdateMountTableEntryResponse response = getMountTableStore()
.updateMountTableEntry(request);
try {
MountTable mountTable = request.getEntry();
if (mountTable != null && router.isQuotaEnabled()
&& isQuotaUpdated(request, mountTable)) {
synchronizeQuota(mountTable.getSourcePath(),
mountTable.getQuota().getQuota(),
mountTable.getQuota().getSpaceQuota());
if (updateEntry != null && router.isQuotaEnabled()
&& isQuotaUpdated(request, oldEntry)) {
synchronizeQuota(updateEntry.getSourcePath(),
updateEntry.getQuota().getQuota(),
updateEntry.getQuota().getSpaceQuota());
}
} catch (Exception e) {
// Ignore exception, if any while reseting quota. Specifically to handle
@ -289,29 +294,41 @@ && isQuotaUpdated(request, mountTable)) {
return response;
}
/**
* Checks whether quota needs to be synchronized with namespace or not. Quota
* needs to be synchronized either if there is change in mount entry quota or
* there is change in remote destinations.
* @param request the update request.
* @param oldEntry the mount entry before getting updated.
* @return true if quota needs to be updated.
* @throws IOException
*/
private boolean isQuotaUpdated(UpdateMountTableEntryRequest request,
MountTable mountTable) throws IOException {
long nsQuota = -1;
long ssQuota = -1;
String path = request.getEntry().getSourcePath();
if (this.router.getSubclusterResolver() instanceof MountTableResolver) {
MountTableResolver mResolver = (MountTableResolver) this.router
.getSubclusterResolver();
MountTable entry = mResolver.getMountPoint(path);
if (entry != null) {
RouterQuotaUsage preQuota = entry.getQuota();
nsQuota = preQuota.getQuota();
ssQuota = preQuota.getSpaceQuota();
MountTable oldEntry) throws IOException {
if (oldEntry != null) {
MountTable updateEntry = request.getEntry();
// If locations are changed, the new destinations need to be in sync with
// the mount quota.
if (!oldEntry.getDestinations().equals(updateEntry.getDestinations())) {
return true;
}
}
RouterQuotaUsage mountQuota = mountTable.getQuota();
if (nsQuota != mountQuota.getQuota()
|| ssQuota != mountQuota.getSpaceQuota()) {
// Previous quota.
RouterQuotaUsage preQuota = oldEntry.getQuota();
long nsQuota = preQuota.getQuota();
long ssQuota = preQuota.getSpaceQuota();
// New quota
RouterQuotaUsage mountQuota = updateEntry.getQuota();
// If there is change in quota, the new quota needs to be synchronized.
if (nsQuota != mountQuota.getQuota()
|| ssQuota != mountQuota.getSpaceQuota()) {
return true;
}
return false;
} else {
// If old entry is not available, sync quota always, since we can't
// conclude no change in quota.
return true;
}
return false;
}
/**
@ -323,15 +340,30 @@ private boolean isQuotaUpdated(UpdateMountTableEntryRequest request,
*/
private void synchronizeQuota(String path, long nsQuota, long ssQuota)
throws IOException {
if (router.isQuotaEnabled() &&
(nsQuota != HdfsConstants.QUOTA_DONT_SET
|| ssQuota != HdfsConstants.QUOTA_DONT_SET)) {
HdfsFileStatus ret = this.router.getRpcServer().getFileInfo(path);
if (ret != null) {
this.router.getRpcServer().getQuotaModule().setQuota(path, nsQuota,
ssQuota, null);
if (isQuotaSyncRequired(nsQuota, ssQuota)) {
if (iStateStoreCache) {
((StateStoreCache) this.router.getSubclusterResolver()).loadCache(true);
}
Quota routerQuota = this.router.getRpcServer().getQuotaModule();
routerQuota.setQuota(path, nsQuota, ssQuota, null);
}
}
/**
* Checks if quota needs to be synchronized or not.
* @param nsQuota namespace quota to be set.
* @param ssQuota space quota to be set.
* @return true if the quota needs to be synchronized.
*/
private boolean isQuotaSyncRequired(long nsQuota, long ssQuota) {
// Check if quota is enabled for router or not.
if (router.isQuotaEnabled()) {
if ((nsQuota != HdfsConstants.QUOTA_DONT_SET
|| ssQuota != HdfsConstants.QUOTA_DONT_SET)) {
return true;
}
}
return false;
}
@Override

9
hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterQuota.java
View File

@ -52,6 +52,7 @@
import org.apache.hadoop.hdfs.server.federation.StateStoreDFSCluster;
import org.apache.hadoop.hdfs.server.federation.resolver.MountTableManager;
import org.apache.hadoop.hdfs.server.federation.resolver.MountTableResolver;
import org.apache.hadoop.hdfs.server.federation.resolver.RemoteLocation;
import org.apache.hadoop.hdfs.server.federation.store.protocol.AddMountTableEntryRequest;
import org.apache.hadoop.hdfs.server.federation.store.protocol.AddMountTableEntryResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesRequest;
@ -474,6 +475,14 @@ public void testQuotaUpdating() throws Exception {
assertEquals(ssQuota, quota.getSpaceQuota());
assertEquals(3, quota.getFileAndDirectoryCount());
assertEquals(BLOCK_SIZE, quota.getSpaceConsumed());
// verify quota sync on adding new destination to mount entry.
updatedMountTable = getMountTable(path);
nnFs1.mkdirs(new Path("/newPath"));
updatedMountTable.setDestinations(
Collections.singletonList(new RemoteLocation("ns0", "/newPath", path)));
updateMountTable(updatedMountTable);
assertEquals(nsQuota, nnFs1.getQuotaUsage(new Path("/newPath")).getQuota());
}
/**