You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by in...@apache.org on 2019/03/03 18:37:20 UTC
[hadoop] 41/45: HDFS-14249. RBF: Tooling to identify the subcluster
location of a file. Contributed by Inigo Goiri.
This is an automated email from the ASF dual-hosted git repository.
inigoiri pushed a commit to branch HDFS-13891
in repository https://gitbox.apache.org/repos/asf/hadoop.git
commit 0bbe324c5a752bbc0692134170f442071027ff22
Author: Giovanni Matteo Fumarola <gi...@apache.org>
AuthorDate: Wed Feb 20 11:08:55 2019 -0800
HDFS-14249. RBF: Tooling to identify the subcluster location of a file. Contributed by Inigo Goiri.
---
.../RouterAdminProtocolServerSideTranslatorPB.java | 22 ++++
.../RouterAdminProtocolTranslatorPB.java | 21 +++
.../metrics/FederationRPCPerformanceMonitor.java | 8 +-
.../federation/resolver/MountTableManager.java | 12 ++
.../federation/router/RouterAdminServer.java | 36 ++++++
.../federation/store/impl/MountTableStoreImpl.java | 7 +
.../store/protocol/GetDestinationRequest.java | 57 ++++++++
.../store/protocol/GetDestinationResponse.java | 59 +++++++++
.../impl/pb/GetDestinationRequestPBImpl.java | 73 +++++++++++
.../impl/pb/GetDestinationResponsePBImpl.java | 83 ++++++++++++
.../hadoop/hdfs/tools/federation/RouterAdmin.java | 28 +++-
.../src/main/proto/FederationProtocol.proto | 8 ++
.../src/main/proto/RouterProtocol.proto | 5 +
.../src/site/markdown/HDFSRouterFederation.md | 4 +
.../federation/router/TestRouterAdminCLI.java | 64 ++++++++-
...erRPCMultipleDestinationMountTableResolver.java | 144 +++++++++++++++++++++
.../hadoop-hdfs/src/site/markdown/HDFSCommands.md | 2 +
17 files changed, 628 insertions(+), 5 deletions(-)
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/protocolPB/RouterAdminProtocolServerSideTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/protocolPB/RouterAdminProtocolServerSideTranslatorPB.java
index a31c46d..6f6724e 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/protocolPB/RouterAdminProtocolServerSideTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/protocolPB/RouterAdminProtocolServerSideTranslatorPB.java
@@ -31,6 +31,8 @@ import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProt
import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.EnterSafeModeResponseProto;
import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetDisabledNameservicesRequestProto;
import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetDisabledNameservicesResponseProto;
+import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetDestinationRequestProto;
+import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetDestinationResponseProto;
import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetMountTableEntriesRequestProto;
import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetMountTableEntriesResponseProto;
import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetSafeModeRequestProto;
@@ -54,6 +56,8 @@ import org.apache.hadoop.hdfs.server.federation.store.protocol.EnterSafeModeRequ
import org.apache.hadoop.hdfs.server.federation.store.protocol.EnterSafeModeResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDisabledNameservicesRequest;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDisabledNameservicesResponse;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDestinationRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDestinationResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesRequest;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetSafeModeRequest;
@@ -76,6 +80,8 @@ import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.EnterSafe
import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.EnterSafeModeResponsePBImpl;
import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.GetDisabledNameservicesRequestPBImpl;
import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.GetDisabledNameservicesResponsePBImpl;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.GetDestinationRequestPBImpl;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.GetDestinationResponsePBImpl;
import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.GetMountTableEntriesRequestPBImpl;
import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.GetMountTableEntriesResponsePBImpl;
import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.GetSafeModeRequestPBImpl;
@@ -298,4 +304,20 @@ public class RouterAdminProtocolServerSideTranslatorPB implements
throw new ServiceException(e);
}
}
+
+ @Override
+ public GetDestinationResponseProto getDestination(
+ RpcController controller, GetDestinationRequestProto request)
+ throws ServiceException {
+ try {
+ GetDestinationRequest req =
+ new GetDestinationRequestPBImpl(request);
+ GetDestinationResponse response = server.getDestination(req);
+ GetDestinationResponsePBImpl responsePB =
+ (GetDestinationResponsePBImpl)response;
+ return responsePB.getProto();
+ } catch (IOException e) {
+ throw new ServiceException(e);
+ }
+ }
}
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/protocolPB/RouterAdminProtocolTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/protocolPB/RouterAdminProtocolTranslatorPB.java
index 1fbb06d..9cdc3c1 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/protocolPB/RouterAdminProtocolTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/protocolPB/RouterAdminProtocolTranslatorPB.java
@@ -32,6 +32,8 @@ import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProt
import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.EnterSafeModeResponseProto;
import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetDisabledNameservicesRequestProto;
import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetDisabledNameservicesResponseProto;
+import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetDestinationRequestProto;
+import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetDestinationResponseProto;
import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetMountTableEntriesRequestProto;
import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetMountTableEntriesResponseProto;
import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetSafeModeRequestProto;
@@ -57,6 +59,8 @@ import org.apache.hadoop.hdfs.server.federation.store.protocol.EnterSafeModeRequ
import org.apache.hadoop.hdfs.server.federation.store.protocol.EnterSafeModeResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDisabledNameservicesRequest;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDisabledNameservicesResponse;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDestinationRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDestinationResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesRequest;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetSafeModeRequest;
@@ -77,6 +81,8 @@ import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.EnableNam
import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.EnableNameserviceResponsePBImpl;
import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.EnterSafeModeResponsePBImpl;
import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.GetDisabledNameservicesResponsePBImpl;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.GetDestinationRequestPBImpl;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.GetDestinationResponsePBImpl;
import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.GetMountTableEntriesRequestPBImpl;
import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.GetMountTableEntriesResponsePBImpl;
import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.GetSafeModeResponsePBImpl;
@@ -288,4 +294,19 @@ public class RouterAdminProtocolTranslatorPB
throw new IOException(ProtobufHelper.getRemoteException(e).getMessage());
}
}
+
+ @Override
+ public GetDestinationResponse getDestination(
+ GetDestinationRequest request) throws IOException {
+ GetDestinationRequestPBImpl requestPB =
+ (GetDestinationRequestPBImpl) request;
+ GetDestinationRequestProto proto = requestPB.getProto();
+ try {
+ GetDestinationResponseProto response =
+ rpcProxy.getDestination(null, proto);
+ return new GetDestinationResponsePBImpl(response);
+ } catch (ServiceException e) {
+ throw new IOException(ProtobufHelper.getRemoteException(e).getMessage());
+ }
+ }
}
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/metrics/FederationRPCPerformanceMonitor.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/metrics/FederationRPCPerformanceMonitor.java
index cbd63de..bae83aa 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/metrics/FederationRPCPerformanceMonitor.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/metrics/FederationRPCPerformanceMonitor.java
@@ -129,7 +129,7 @@ public class FederationRPCPerformanceMonitor implements RouterRpcMonitor {
public long proxyOp() {
PROXY_TIME.set(monotonicNow());
long processingTime = getProcessingTime();
- if (processingTime >= 0) {
+ if (metrics != null && processingTime >= 0) {
metrics.addProcessingTime(processingTime);
}
return Thread.currentThread().getId();
@@ -139,7 +139,7 @@ public class FederationRPCPerformanceMonitor implements RouterRpcMonitor {
public void proxyOpComplete(boolean success) {
if (success) {
long proxyTime = getProxyTime();
- if (proxyTime >= 0) {
+ if (metrics != null && proxyTime >= 0) {
metrics.addProxyTime(proxyTime);
}
}
@@ -147,7 +147,9 @@ public class FederationRPCPerformanceMonitor implements RouterRpcMonitor {
@Override
public void proxyOpFailureStandby() {
- metrics.incrProxyOpFailureStandby();
+ if (metrics != null) {
+ metrics.incrProxyOpFailureStandby();
+ }
}
@Override
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MountTableManager.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MountTableManager.java
index 9a1e416..5ff2e28 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MountTableManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MountTableManager.java
@@ -21,6 +21,8 @@ import java.io.IOException;
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.GetDestinationRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDestinationResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesRequest;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.RefreshMountTableEntriesRequest;
@@ -93,4 +95,14 @@ public interface MountTableManager {
*/
RefreshMountTableEntriesResponse refreshMountTableEntries(
RefreshMountTableEntriesRequest request) throws IOException;
+
+ /**
+ * Get the destination subcluster (namespace) of a file/directory.
+ *
+ * @param request Fully populated request object including the file to check.
+ * @return The response including the subcluster where the input file is.
+ * @throws IOException Throws exception if the data store is not initialized.
+ */
+ GetDestinationResponse getDestination(
+ GetDestinationRequest request) throws IOException;
}
\ No newline at end of file
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterAdminServer.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterAdminServer.java
index e2d944c..a2a5a42 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterAdminServer.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterAdminServer.java
@@ -23,7 +23,10 @@ import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_PERMISSIONS_ENABLED_KEY;
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.List;
+import java.util.Map;
import java.util.Set;
import com.google.common.base.Preconditions;
@@ -39,6 +42,7 @@ import org.apache.hadoop.hdfs.protocolPB.RouterAdminProtocolServerSideTranslator
import org.apache.hadoop.hdfs.protocolPB.RouterPolicyProvider;
import org.apache.hadoop.hdfs.server.federation.resolver.ActiveNamenodeResolver;
import org.apache.hadoop.hdfs.server.federation.resolver.FederationNamespaceInfo;
+import org.apache.hadoop.hdfs.server.federation.resolver.RemoteLocation;
import org.apache.hadoop.hdfs.server.federation.store.DisabledNameserviceStore;
import org.apache.hadoop.hdfs.server.federation.store.MountTableStore;
import org.apache.hadoop.hdfs.server.federation.store.StateStoreCache;
@@ -52,6 +56,8 @@ import org.apache.hadoop.hdfs.server.federation.store.protocol.EnterSafeModeRequ
import org.apache.hadoop.hdfs.server.federation.store.protocol.EnterSafeModeResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDisabledNameservicesRequest;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDisabledNameservicesResponse;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDestinationRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDestinationResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesRequest;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetSafeModeRequest;
@@ -378,6 +384,36 @@ public class RouterAdminServer extends AbstractService
}
}
+ @Override
+ public GetDestinationResponse getDestination(
+ GetDestinationRequest request) throws IOException {
+ final String src = request.getSrcPath();
+ final List<String> nsIds = new ArrayList<>();
+ RouterRpcServer rpcServer = this.router.getRpcServer();
+ List<RemoteLocation> locations = rpcServer.getLocationsForPath(src, false);
+ RouterRpcClient rpcClient = rpcServer.getRPCClient();
+ RemoteMethod method = new RemoteMethod("getFileInfo",
+ new Class<?>[] {String.class}, new RemoteParam());
+ try {
+ Map<RemoteLocation, HdfsFileStatus> responses =
+ rpcClient.invokeConcurrent(
+ locations, method, false, false, HdfsFileStatus.class);
+ for (RemoteLocation location : locations) {
+ if (responses.get(location) != null) {
+ nsIds.add(location.getNameserviceId());
+ }
+ }
+ } catch (IOException ioe) {
+ LOG.error("Cannot get location for {}: {}",
+ src, ioe.getMessage());
+ }
+ if (nsIds.isEmpty() && !locations.isEmpty()) {
+ String nsId = locations.get(0).getNameserviceId();
+ nsIds.add(nsId);
+ }
+ return GetDestinationResponse.newInstance(nsIds);
+ }
+
/**
* Verify if Router set safe mode state correctly.
* @param isInSafeMode Expected state to be set.
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MountTableStoreImpl.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MountTableStoreImpl.java
index 76c7e78..d5e1857 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MountTableStoreImpl.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MountTableStoreImpl.java
@@ -31,6 +31,8 @@ import org.apache.hadoop.hdfs.server.federation.store.MountTableStore;
import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver;
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.GetDestinationRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDestinationResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesRequest;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.RefreshMountTableEntriesRequest;
@@ -169,4 +171,9 @@ public class MountTableStoreImpl extends MountTableStore {
return response;
}
+ @Override
+ public GetDestinationResponse getDestination(
+ GetDestinationRequest request) throws IOException {
+ throw new UnsupportedOperationException("Requires the RouterRpcServer");
+ }
}
\ No newline at end of file
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/GetDestinationRequest.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/GetDestinationRequest.java
new file mode 100644
index 0000000..0d5074b
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/GetDestinationRequest.java
@@ -0,0 +1,57 @@
+/**
+ * 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.federation.store.protocol;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience.Public;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreSerializer;
+
+/**
+ * API request for getting the destination subcluster of a file.
+ */
+public abstract class GetDestinationRequest {
+
+ public static GetDestinationRequest newInstance()
+ throws IOException {
+ return StateStoreSerializer
+ .newRecord(GetDestinationRequest.class);
+ }
+
+ public static GetDestinationRequest newInstance(String srcPath)
+ throws IOException {
+ GetDestinationRequest request = newInstance();
+ request.setSrcPath(srcPath);
+ return request;
+ }
+
+ public static GetDestinationRequest newInstance(Path srcPath)
+ throws IOException {
+ return newInstance(srcPath.toString());
+ }
+
+ @Public
+ @Unstable
+ public abstract String getSrcPath();
+
+ @Public
+ @Unstable
+ public abstract void setSrcPath(String srcPath);
+}
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/GetDestinationResponse.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/GetDestinationResponse.java
new file mode 100644
index 0000000..534b673
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/GetDestinationResponse.java
@@ -0,0 +1,59 @@
+/**
+ * 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.federation.store.protocol;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
+
+import org.apache.hadoop.classification.InterfaceAudience.Public;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreSerializer;
+
+/**
+ * API response for getting the destination subcluster of a file.
+ */
+public abstract class GetDestinationResponse {
+
+ public static GetDestinationResponse newInstance()
+ throws IOException {
+ return StateStoreSerializer
+ .newRecord(GetDestinationResponse.class);
+ }
+
+ public static GetDestinationResponse newInstance(
+ Collection<String> nsIds) throws IOException {
+ GetDestinationResponse request = newInstance();
+ request.setDestinations(nsIds);
+ return request;
+ }
+
+ @Public
+ @Unstable
+ public abstract Collection<String> getDestinations();
+
+ @Public
+ @Unstable
+ public void setDestination(String nsId) {
+ setDestinations(Collections.singletonList(nsId));
+ }
+
+ @Public
+ @Unstable
+ public abstract void setDestinations(Collection<String> nsIds);
+}
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/impl/pb/GetDestinationRequestPBImpl.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/impl/pb/GetDestinationRequestPBImpl.java
new file mode 100644
index 0000000..b97f455
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/impl/pb/GetDestinationRequestPBImpl.java
@@ -0,0 +1,73 @@
+/**
+ * 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.federation.store.protocol.impl.pb;
+
+import java.io.IOException;
+
+import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetDestinationRequestProto;
+import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetDestinationRequestProtoOrBuilder;
+import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetDestinationRequestProto.Builder;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDestinationRequest;
+import org.apache.hadoop.hdfs.server.federation.store.records.impl.pb.PBRecord;
+
+import com.google.protobuf.Message;
+
+/**
+ * Protobuf implementation of the state store API object
+ * GetDestinationRequest.
+ */
+public class GetDestinationRequestPBImpl extends GetDestinationRequest
+ implements PBRecord {
+
+ private FederationProtocolPBTranslator<GetDestinationRequestProto,
+ Builder, GetDestinationRequestProtoOrBuilder> translator =
+ new FederationProtocolPBTranslator<>(
+ GetDestinationRequestProto.class);
+
+ public GetDestinationRequestPBImpl() {
+ }
+
+ public GetDestinationRequestPBImpl(GetDestinationRequestProto proto) {
+ this.translator.setProto(proto);
+ }
+
+ @Override
+ public GetDestinationRequestProto getProto() {
+ return this.translator.build();
+ }
+
+ @Override
+ public void setProto(Message proto) {
+ this.translator.setProto(proto);
+ }
+
+ @Override
+ public void readInstance(String base64String) throws IOException {
+ this.translator.readInstance(base64String);
+ }
+
+ @Override
+ public String getSrcPath() {
+ return this.translator.getProtoOrBuilder().getSrcPath();
+ }
+
+ @Override
+ public void setSrcPath(String path) {
+ this.translator.getBuilder().setSrcPath(path);
+ }
+}
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/impl/pb/GetDestinationResponsePBImpl.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/impl/pb/GetDestinationResponsePBImpl.java
new file mode 100644
index 0000000..f758f99
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/impl/pb/GetDestinationResponsePBImpl.java
@@ -0,0 +1,83 @@
+/**
+ * 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.federation.store.protocol.impl.pb;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+
+import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetDestinationResponseProto;
+import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetDestinationResponseProto.Builder;
+import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.GetDestinationResponseProtoOrBuilder;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDestinationResponse;
+import org.apache.hadoop.hdfs.server.federation.store.records.impl.pb.PBRecord;
+
+import com.google.protobuf.Message;
+
+/**
+ * Protobuf implementation of the state store API object
+ * GetDestinationResponse.
+ */
+public class GetDestinationResponsePBImpl
+ extends GetDestinationResponse implements PBRecord {
+
+ private FederationProtocolPBTranslator<GetDestinationResponseProto,
+ Builder, GetDestinationResponseProtoOrBuilder> translator =
+ new FederationProtocolPBTranslator<>(
+ GetDestinationResponseProto.class);
+
+ public GetDestinationResponsePBImpl() {
+ }
+
+ public GetDestinationResponsePBImpl(
+ GetDestinationResponseProto proto) {
+ this.translator.setProto(proto);
+ }
+
+ @Override
+ public GetDestinationResponseProto getProto() {
+ // if builder is null build() returns null, calling getBuilder() to
+ // instantiate builder
+ this.translator.getBuilder();
+ return this.translator.build();
+ }
+
+ @Override
+ public void setProto(Message proto) {
+ this.translator.setProto(proto);
+ }
+
+ @Override
+ public void readInstance(String base64String) throws IOException {
+ this.translator.readInstance(base64String);
+ }
+
+ @Override
+ public Collection<String> getDestinations() {
+ return new ArrayList<>(
+ this.translator.getProtoOrBuilder().getDestinationsList());
+ }
+
+ @Override
+ public void setDestinations(Collection<String> nsIds) {
+ this.translator.getBuilder().clearDestinations();
+ for (String nsId : nsIds) {
+ this.translator.getBuilder().addDestinations(nsId);
+ }
+ }
+}
\ No newline at end of file
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/tools/federation/RouterAdmin.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/tools/federation/RouterAdmin.java
index 37aad88..b04b069 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/tools/federation/RouterAdmin.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/tools/federation/RouterAdmin.java
@@ -52,6 +52,8 @@ import org.apache.hadoop.hdfs.server.federation.store.protocol.EnterSafeModeRequ
import org.apache.hadoop.hdfs.server.federation.store.protocol.EnterSafeModeResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDisabledNameservicesRequest;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDisabledNameservicesResponse;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDestinationRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDestinationResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesRequest;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetSafeModeRequest;
@@ -117,7 +119,8 @@ public class RouterAdmin extends Configured implements Tool {
private String getUsage(String cmd) {
if (cmd == null) {
String[] commands =
- {"-add", "-update", "-rm", "-ls", "-setQuota", "-clrQuota",
+ {"-add", "-update", "-rm", "-ls", "-getDestination",
+ "-setQuota", "-clrQuota",
"-safemode", "-nameservice", "-getDisabledNameservices",
"-refresh"};
StringBuilder usage = new StringBuilder();
@@ -143,6 +146,8 @@ public class RouterAdmin extends Configured implements Tool {
return "\t[-rm <source>]";
} else if (cmd.equals("-ls")) {
return "\t[-ls <path>]";
+ } else if (cmd.equals("-getDestination")) {
+ return "\t[-getDestination <path>]";
} else if (cmd.equals("-setQuota")) {
return "\t[-setQuota <path> -nsQuota <nsQuota> -ssQuota "
+ "<quota in bytes or quota size string>]";
@@ -172,6 +177,11 @@ public class RouterAdmin extends Configured implements Tool {
throw new IllegalArgumentException(
"Too many arguments, Max=1 argument allowed");
}
+ } else if (arg[0].equals("-getDestination")) {
+ if (arg.length > 2) {
+ throw new IllegalArgumentException(
+ "Too many arguments, Max=1 argument allowed only");
+ }
} else if (arg[0].equals("-safemode")) {
if (arg.length > 2) {
throw new IllegalArgumentException(
@@ -208,6 +218,10 @@ public class RouterAdmin extends Configured implements Tool {
if (argv.length < 2) {
return false;
}
+ } else if ("-getDestination".equals(cmd)) {
+ if (argv.length < 2) {
+ return false;
+ }
} else if ("-setQuota".equals(cmd)) {
if (argv.length < 4) {
return false;
@@ -302,6 +316,8 @@ public class RouterAdmin extends Configured implements Tool {
} else {
listMounts("/");
}
+ } else if ("-getDestination".equals(cmd)) {
+ getDestination(argv[i]);
} else if ("-setQuota".equals(cmd)) {
if (setQuota(argv, i)) {
System.out.println(
@@ -709,6 +725,16 @@ public class RouterAdmin extends Configured implements Tool {
}
}
+ private void getDestination(String path) throws IOException {
+ path = normalizeFileSystemPath(path);
+ MountTableManager mountTable = client.getMountTableManager();
+ GetDestinationRequest request =
+ GetDestinationRequest.newInstance(path);
+ GetDestinationResponse response = mountTable.getDestination(request);
+ System.out.println("Destination: " +
+ StringUtils.join(",", response.getDestinations()));
+ }
+
/**
* Set quota for a mount table entry.
*
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/proto/FederationProtocol.proto b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/proto/FederationProtocol.proto
index 1e5e37b..9e9fd48 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/proto/FederationProtocol.proto
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/proto/FederationProtocol.proto
@@ -175,6 +175,14 @@ message GetMountTableEntriesResponseProto {
optional uint64 timestamp = 2;
}
+message GetDestinationRequestProto {
+ optional string srcPath = 1;
+}
+
+message GetDestinationResponseProto {
+ repeated string destinations = 1;
+}
+
/////////////////////////////////////////////////
// Routers
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/proto/RouterProtocol.proto b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/proto/RouterProtocol.proto
index 34a012a..d6aff49 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/proto/RouterProtocol.proto
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/proto/RouterProtocol.proto
@@ -79,4 +79,9 @@ service RouterAdminProtocolService {
* Refresh mount entries
*/
rpc refreshMountTableEntries(RefreshMountTableEntriesRequestProto) returns(RefreshMountTableEntriesResponseProto);
+
+ /**
+ * Get the destination of a file/directory in the federation.
+ */
+ rpc getDestination(GetDestinationRequestProto) returns (GetDestinationResponseProto);
}
\ No newline at end of file
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/site/markdown/HDFSRouterFederation.md b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/site/markdown/HDFSRouterFederation.md
index 2ae0c2b..f24ff12 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/site/markdown/HDFSRouterFederation.md
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/site/markdown/HDFSRouterFederation.md
@@ -261,6 +261,10 @@ RANDOM can be used for reading and writing data from/into different subclusters.
The common use for this approach is to have the same data in multiple subclusters and balance the reads across subclusters.
For example, if thousands of containers need to read the same data (e.g., a library), one can use RANDOM to read the data from any of the subclusters.
+To determine which subcluster contains a file:
+
+ [hdfs]$ $HADOOP_HOME/bin/hdfs dfsrouteradmin -getDestination /user/user1/file.txt
+
Note that consistency of the data across subclusters is not guaranteed by the Router.
### Disabling nameservices
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAdminCLI.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAdminCLI.java
index ab733dd..9f53dd4 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAdminCLI.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAdminCLI.java
@@ -26,6 +26,11 @@ import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.net.InetSocketAddress;
import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
@@ -36,6 +41,8 @@ import org.apache.hadoop.hdfs.server.federation.StateStoreDFSCluster;
import org.apache.hadoop.hdfs.server.federation.metrics.FederationMetrics;
import org.apache.hadoop.hdfs.server.federation.resolver.ActiveNamenodeResolver;
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.MultipleDestinationMountTableResolver;
import org.apache.hadoop.hdfs.server.federation.resolver.RemoteLocation;
import org.apache.hadoop.hdfs.server.federation.resolver.order.DestinationOrder;
import org.apache.hadoop.hdfs.server.federation.store.StateStoreService;
@@ -78,7 +85,8 @@ public class TestRouterAdminCLI {
@BeforeClass
public static void globalSetUp() throws Exception {
- cluster = new StateStoreDFSCluster(false, 1);
+ cluster = new StateStoreDFSCluster(false, 1,
+ MultipleDestinationMountTableResolver.class);
// Build and start a router with State Store + admin + RPC
Configuration conf = new RouterConfigBuilder()
.stateStore()
@@ -550,6 +558,11 @@ public class TestRouterAdminCLI {
.contains("\t[-nameservice enable | disable <nameservice>]"));
out.reset();
+ argv = new String[] {"-getDestination"};
+ assertEquals(-1, ToolRunner.run(admin, argv));
+ assertTrue(out.toString().contains("\t[-getDestination <path>]"));
+ out.reset();
+
argv = new String[] {"-Random"};
assertEquals(-1, ToolRunner.run(admin, argv));
String expected = "Usage: hdfs dfsrouteradmin :\n"
@@ -560,6 +573,7 @@ public class TestRouterAdminCLI {
+ "<destination> " + "[-readonly] [-order HASH|LOCAL|RANDOM|HASH_ALL] "
+ "-owner <owner> -group <group> -mode <mode>]\n" + "\t[-rm <source>]\n"
+ "\t[-ls <path>]\n"
+ + "\t[-getDestination <path>]\n"
+ "\t[-setQuota <path> -nsQuota <nsQuota> -ssQuota "
+ "<quota in bytes or quota size string>]\n" + "\t[-clrQuota <path>]\n"
+ "\t[-safemode enter | leave | get]\n"
@@ -1091,4 +1105,52 @@ public class TestRouterAdminCLI {
assertEquals(dest, mountTable.getDestinations().get(0).getDest());
assertEquals(order, mountTable.getDestOrder());
}
+
+ @Test
+ public void testGetDestination() throws Exception {
+
+ // Test the basic destination feature
+ System.setOut(new PrintStream(out));
+ String[] argv = new String[] {"-getDestination", "/file.txt"};
+ assertEquals(0, ToolRunner.run(admin, argv));
+ assertEquals("Destination: ns0" + System.lineSeparator(), out.toString());
+
+ // Add a HASH_ALL entry to check the destination changing
+ argv = new String[] {"-add", "/testGetDest", "ns0,ns1",
+ "/testGetDestination",
+ "-order", DestinationOrder.HASH_ALL.toString()};
+ assertEquals(0, ToolRunner.run(admin, argv));
+ stateStore.loadCache(MountTableStoreImpl.class, true);
+ MountTableResolver resolver =
+ (MountTableResolver) router.getSubclusterResolver();
+ resolver.loadCache(true);
+
+ // Files should be distributed across ns0 and ns1
+ Map<String, AtomicInteger> counter = new TreeMap<>();
+ final Pattern p = Pattern.compile("Destination: (.*)");
+ for (int i = 0; i < 10; i++) {
+ out.reset();
+ String filename = "file" + i+ ".txt";
+ argv = new String[] {"-getDestination", "/testGetDest/" + filename};
+ assertEquals(0, ToolRunner.run(admin, argv));
+ String outLine = out.toString();
+ Matcher m = p.matcher(outLine);
+ assertTrue(m.find());
+ String nsId = m.group(1);
+ if (counter.containsKey(nsId)) {
+ counter.get(nsId).getAndIncrement();
+ } else {
+ counter.put(nsId, new AtomicInteger(1));
+ }
+ }
+ assertEquals("Wrong counter size: " + counter, 2, counter.size());
+ assertTrue(counter + " should contain ns0", counter.containsKey("ns0"));
+ assertTrue(counter + " should contain ns1", counter.containsKey("ns1"));
+
+ // Bad cases
+ argv = new String[] {"-getDestination"};
+ assertEquals(-1, ToolRunner.run(admin, argv));
+ argv = new String[] {"-getDestination /file1.txt /file2.txt"};
+ assertEquals(-1, ToolRunner.run(admin, argv));
+ }
}
\ No newline at end of file
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterRPCMultipleDestinationMountTableResolver.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterRPCMultipleDestinationMountTableResolver.java
index 8c15151..46bfff9 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterRPCMultipleDestinationMountTableResolver.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterRPCMultipleDestinationMountTableResolver.java
@@ -23,11 +23,19 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
+import java.io.FileNotFoundException;
import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
+import java.util.TreeSet;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.DFSTestUtil;
@@ -41,8 +49,11 @@ import org.apache.hadoop.hdfs.server.federation.resolver.MultipleDestinationMoun
import org.apache.hadoop.hdfs.server.federation.resolver.order.DestinationOrder;
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.GetDestinationRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetDestinationResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.RemoveMountTableEntryRequest;
import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
+import org.apache.hadoop.test.LambdaTestUtils;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.BeforeClass;
@@ -52,6 +63,8 @@ import org.junit.Test;
* Tests router rpc with multiple destination mount table resolver.
*/
public class TestRouterRPCMultipleDestinationMountTableResolver {
+ private static final List<String> NS_IDS = Arrays.asList("ns0", "ns1");
+
private static StateStoreDFSCluster cluster;
private static RouterContext routerContext;
private static MountTableResolver resolver;
@@ -391,4 +404,135 @@ public class TestRouterRPCMultipleDestinationMountTableResolver {
return addResponse.getStatus();
}
+
+ @Test
+ public void testGetDestinationHashAll() throws Exception {
+ testGetDestination(DestinationOrder.HASH_ALL,
+ Arrays.asList("ns1"),
+ Arrays.asList("ns1"),
+ Arrays.asList("ns1", "ns0"));
+ }
+
+ @Test
+ public void testGetDestinationHash() throws Exception {
+ testGetDestination(DestinationOrder.HASH,
+ Arrays.asList("ns1"),
+ Arrays.asList("ns1"),
+ Arrays.asList("ns1"));
+ }
+
+ @Test
+ public void testGetDestinationRandom() throws Exception {
+ testGetDestination(DestinationOrder.RANDOM,
+ null, null, Arrays.asList("ns0", "ns1"));
+ }
+
+ /**
+ * Generic test for getting the destination subcluster.
+ * @param order DestinationOrder of the mount point.
+ * @param expectFileLocation Expected subclusters of a file. null for any.
+ * @param expectNoFileLocation Expected subclusters of a non-existing file.
+ * @param expectDirLocation Expected subclusters of a nested directory.
+ * @throws Exception If the test cannot run.
+ */
+ private void testGetDestination(DestinationOrder order,
+ List<String> expectFileLocation,
+ List<String> expectNoFileLocation,
+ List<String> expectDirLocation) throws Exception {
+ setupOrderMountPath(order);
+
+ RouterClient client = routerContext.getAdminClient();
+ MountTableManager mountTableManager = client.getMountTableManager();
+
+ // If the file exists, it should be in the expected subcluster
+ final String pathFile = "dir/file";
+ final Path pathRouterFile = new Path("/mount", pathFile);
+ final Path pathLocalFile = new Path("/tmp", pathFile);
+ FileStatus fileStatus = routerFs.getFileStatus(pathRouterFile);
+ assertTrue(fileStatus + " should be a file", fileStatus.isFile());
+ GetDestinationResponse respFile = mountTableManager.getDestination(
+ GetDestinationRequest.newInstance(pathRouterFile));
+ if (expectFileLocation != null) {
+ assertEquals(expectFileLocation, respFile.getDestinations());
+ assertPathStatus(expectFileLocation, pathLocalFile, false);
+ } else {
+ Collection<String> dests = respFile.getDestinations();
+ assertPathStatus(dests, pathLocalFile, false);
+ }
+
+ // If the file does not exist, it should give us the expected subclusters
+ final String pathNoFile = "dir/no-file";
+ final Path pathRouterNoFile = new Path("/mount", pathNoFile);
+ final Path pathLocalNoFile = new Path("/tmp", pathNoFile);
+ LambdaTestUtils.intercept(FileNotFoundException.class,
+ () -> routerFs.getFileStatus(pathRouterNoFile));
+ GetDestinationResponse respNoFile = mountTableManager.getDestination(
+ GetDestinationRequest.newInstance(pathRouterNoFile));
+ if (expectNoFileLocation != null) {
+ assertEquals(expectNoFileLocation, respNoFile.getDestinations());
+ }
+ assertPathStatus(Collections.emptyList(), pathLocalNoFile, false);
+
+ // If the folder exists, it should be in the expected subcluster
+ final String pathNestedDir = "dir/dir";
+ final Path pathRouterNestedDir = new Path("/mount", pathNestedDir);
+ final Path pathLocalNestedDir = new Path("/tmp", pathNestedDir);
+ FileStatus dirStatus = routerFs.getFileStatus(pathRouterNestedDir);
+ assertTrue(dirStatus + " should be a directory", dirStatus.isDirectory());
+ GetDestinationResponse respDir = mountTableManager.getDestination(
+ GetDestinationRequest.newInstance(pathRouterNestedDir));
+ assertEqualsCollection(expectDirLocation, respDir.getDestinations());
+ assertPathStatus(expectDirLocation, pathLocalNestedDir, true);
+ }
+
+ /**
+ * Assert that the status of a file in the subcluster is the expected one.
+ * @param expectedLocations Subclusters where the file is expected to exist.
+ * @param path Path of the file/directory to check.
+ * @param isDir If the path is expected to be a directory.
+ * @throws Exception If the file cannot be checked.
+ */
+ private void assertPathStatus(Collection<String> expectedLocations,
+ Path path, boolean isDir) throws Exception {
+ for (String nsId : NS_IDS) {
+ final FileSystem fs = getFileSystem(nsId);
+ if (expectedLocations.contains(nsId)) {
+ assertTrue(path + " should exist in " + nsId, fs.exists(path));
+ final FileStatus status = fs.getFileStatus(path);
+ if (isDir) {
+ assertTrue(path + " should be a directory", status.isDirectory());
+ } else {
+ assertTrue(path + " should be a file", status.isFile());
+ }
+ } else {
+ assertFalse(path + " should not exist in " + nsId, fs.exists(path));
+ }
+ }
+ }
+
+ /**
+ * Assert if two collections are equal without checking the order.
+ * @param col1 First collection to compare.
+ * @param col2 Second collection to compare.
+ */
+ private static void assertEqualsCollection(
+ Collection<String> col1, Collection<String> col2) {
+ assertEquals(new TreeSet<>(col1), new TreeSet<>(col2));
+ }
+
+ /**
+ * Get the filesystem for each subcluster.
+ * @param nsId Identifier of the name space (subcluster).
+ * @return The FileSystem for
+ */
+ private static FileSystem getFileSystem(final String nsId) {
+ if (nsId.equals("ns0")) {
+ return nnFs0;
+ }
+ if (nsId.equals("ns1")) {
+ return nnFs1;
+ }
+ return null;
+ }
+
}
\ No newline at end of file
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/site/markdown/HDFSCommands.md b/hadoop-hdfs-project/hadoop-hdfs/src/site/markdown/HDFSCommands.md
index c3f113d..32e88a2 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/site/markdown/HDFSCommands.md
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/site/markdown/HDFSCommands.md
@@ -432,6 +432,7 @@ Usage:
[-update <source> <nameservice1, nameservice2, ...> <destination> [-readonly] [-order HASH|LOCAL|RANDOM|HASH_ALL] -owner <owner> -group <group> -mode <mode>]
[-rm <source>]
[-ls <path>]
+ [-getDestination <path>]
[-setQuota <path> -nsQuota <nsQuota> -ssQuota <quota in bytes or quota size string>]
[-clrQuota <path>]
[-safemode enter | leave | get]
@@ -446,6 +447,7 @@ Usage:
| `-update` *source* *nameservices* *destination* | Update a mount table entry or create one if it does not exist. |
| `-rm` *source* | Remove mount point of specified path. |
| `-ls` *path* | List mount points under specified path. |
+| `-getDestination` *path* | Get the subcluster where a file is or should be created. |
| `-setQuota` *path* `-nsQuota` *nsQuota* `-ssQuota` *ssQuota* | Set quota for specified path. See [HDFS Quotas Guide](./HdfsQuotaAdminGuide.html) for the quota detail. |
| `-clrQuota` *path* | Clear quota of given mount point. See [HDFS Quotas Guide](./HdfsQuotaAdminGuide.html) for the quota detail. |
| `-safemode` `enter` `leave` `get` | Manually set the Router entering or leaving safe mode. The option *get* will be used for verifying if the Router is in safe mode state. |
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org