diff --git a/hadoop-hdfs-project/hadoop-hdfs/pom.xml b/hadoop-hdfs-project/hadoop-hdfs/pom.xml index cc7a975c1a3..93216dbd8ab 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/pom.xml +++ b/hadoop-hdfs-project/hadoop-hdfs/pom.xml @@ -332,6 +332,7 @@ http://maven.apache.org/xsd/maven-4.0.0.xsd"> editlog.proto fsimage.proto FederationProtocol.proto + RouterProtocol.proto diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs b/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs index b1f44a4196e..d51a8e2e3db 100755 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs @@ -31,6 +31,7 @@ function hadoop_usage hadoop_add_option "--hosts filename" "list of hosts to use in worker mode" hadoop_add_option "--workers" "turn on worker mode" +<<<<<<< HEAD hadoop_add_subcommand "balancer" daemon "run a cluster balancing utility" hadoop_add_subcommand "cacheadmin" admin "configure the HDFS cache" hadoop_add_subcommand "classpath" client "prints the class path needed to get the hadoop jar and the required libraries" @@ -42,6 +43,7 @@ function hadoop_usage hadoop_add_subcommand "diskbalancer" daemon "Distributes data evenly among disks on a given node" hadoop_add_subcommand "envvars" client "display computed Hadoop environment variables" hadoop_add_subcommand "ec" admin "run a HDFS ErasureCoding CLI" + hadoop_add_subcommand "federation" admin "manage Router-based federation" hadoop_add_subcommand "fetchdt" client "fetch a delegation token from the NameNode" hadoop_add_subcommand "fsck" admin "run a DFS filesystem checking utility" hadoop_add_subcommand "getconf" client "get config values from configuration" @@ -181,6 +183,9 @@ function hdfscmd_case HADOOP_SUBCMD_SUPPORTDAEMONIZATION="true" HADOOP_CLASSNAME='org.apache.hadoop.hdfs.server.federation.router.Router' ;; + federation) + HADOOP_CLASSNAME='org.apache.hadoop.hdfs.tools.federation.RouterAdmin' + ;; secondarynamenode) HADOOP_SUBCMD_SUPPORTDAEMONIZATION="true" HADOOP_CLASSNAME='org.apache.hadoop.hdfs.server.namenode.SecondaryNameNode' diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs.cmd b/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs.cmd index b9853d622be..53bdf70edac 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs.cmd +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs.cmd @@ -59,7 +59,7 @@ if "%1" == "--loglevel" ( ) ) - set hdfscommands=dfs namenode secondarynamenode journalnode zkfc datanode dfsadmin haadmin fsck balancer jmxget oiv oev fetchdt getconf groups snapshotDiff lsSnapshottableDir cacheadmin mover storagepolicies classpath crypto router debug + set hdfscommands=dfs namenode secondarynamenode journalnode zkfc datanode dfsadmin haadmin fsck balancer jmxget oiv oev fetchdt getconf groups snapshotDiff lsSnapshottableDir cacheadmin mover storagepolicies classpath crypto router federation debug for %%i in ( %hdfscommands% ) do ( if %hdfs-command% == %%i set hdfscommand=true ) @@ -184,6 +184,11 @@ goto :eof set HADOOP_OPTS=%HADOOP_OPTS% %HADOOP_ROUTER_OPTS% goto :eof +:federation + set CLASS=org.apache.hadoop.hdfs.tools.federation.RouterAdmin + set HADOOP_OPTS=%HADOOP_OPTS% %HADOOP_ROUTER_OPTS% + goto :eof + :debug set CLASS=org.apache.hadoop.hdfs.tools.DebugAdmin goto :eof diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java index 9b77a908dba..629ad00fa62 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java @@ -1196,6 +1196,10 @@ public class DFSConfigKeys extends CommonConfigurationKeys { public static final String FEDERATION_STORE_PREFIX = FEDERATION_ROUTER_PREFIX + "store."; + public static final String DFS_ROUTER_STORE_ENABLE = + FEDERATION_STORE_PREFIX + "enable"; + public static final boolean DFS_ROUTER_STORE_ENABLE_DEFAULT = true; + public static final String FEDERATION_STORE_SERIALIZER_CLASS = DFSConfigKeys.FEDERATION_STORE_PREFIX + "serializer"; public static final Class @@ -1222,6 +1226,21 @@ public class DFSConfigKeys extends CommonConfigurationKeys { public static final long FEDERATION_STORE_MEMBERSHIP_EXPIRATION_MS_DEFAULT = TimeUnit.MINUTES.toMillis(5); + // HDFS Router-based federation admin + public static final String DFS_ROUTER_ADMIN_HANDLER_COUNT_KEY = + FEDERATION_ROUTER_PREFIX + "admin.handler.count"; + public static final int DFS_ROUTER_ADMIN_HANDLER_COUNT_DEFAULT = 1; + public static final int DFS_ROUTER_ADMIN_PORT_DEFAULT = 8111; + public static final String DFS_ROUTER_ADMIN_ADDRESS_KEY = + FEDERATION_ROUTER_PREFIX + "admin-address"; + public static final String DFS_ROUTER_ADMIN_ADDRESS_DEFAULT = + "0.0.0.0:" + DFS_ROUTER_ADMIN_PORT_DEFAULT; + public static final String DFS_ROUTER_ADMIN_BIND_HOST_KEY = + FEDERATION_ROUTER_PREFIX + "admin-bind-host"; + public static final String DFS_ROUTER_ADMIN_ENABLE = + FEDERATION_ROUTER_PREFIX + "admin.enable"; + public static final boolean DFS_ROUTER_ADMIN_ENABLE_DEFAULT = true; + // dfs.client.retry confs are moved to HdfsClientConfigKeys.Retry @Deprecated public static final String DFS_CLIENT_RETRY_POLICY_ENABLED_KEY diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/RouterAdminProtocolPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/RouterAdminProtocolPB.java new file mode 100644 index 00000000000..96fa794183c --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/RouterAdminProtocolPB.java @@ -0,0 +1,44 @@ +/** + * 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.protocolPB; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.classification.InterfaceStability; +import org.apache.hadoop.hdfs.DFSConfigKeys; +import org.apache.hadoop.hdfs.protocol.HdfsConstants; +import org.apache.hadoop.hdfs.protocol.proto.RouterProtocolProtos.RouterAdminProtocolService; +import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenSelector; +import org.apache.hadoop.ipc.ProtocolInfo; +import org.apache.hadoop.security.KerberosInfo; +import org.apache.hadoop.security.token.TokenInfo; + +/** + * Protocol that a clients use to communicate with the NameNode. + * Note: This extends the protocolbuffer service based interface to + * add annotations required for security. + */ +@InterfaceAudience.Private +@InterfaceStability.Stable +@KerberosInfo( + serverPrincipal = DFSConfigKeys.DFS_NAMENODE_KERBEROS_PRINCIPAL_KEY) +@TokenInfo(DelegationTokenSelector.class) +@ProtocolInfo(protocolName = HdfsConstants.CLIENT_NAMENODE_PROTOCOL_NAME, + protocolVersion = 1) +public interface RouterAdminProtocolPB extends + RouterAdminProtocolService.BlockingInterface { +} \ No newline at end of file diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/RouterAdminProtocolServerSideTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/RouterAdminProtocolServerSideTranslatorPB.java new file mode 100644 index 00000000000..415bbd9b5c0 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/RouterAdminProtocolServerSideTranslatorPB.java @@ -0,0 +1,151 @@ +/** + * 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.protocolPB; + +import java.io.IOException; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.classification.InterfaceStability; +import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.AddMountTableEntryRequestProto; +import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.AddMountTableEntryResponseProto; +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.RemoveMountTableEntryRequestProto; +import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.RemoveMountTableEntryResponseProto; +import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.UpdateMountTableEntryRequestProto; +import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.UpdateMountTableEntryResponseProto; +import org.apache.hadoop.hdfs.server.federation.router.RouterAdminServer; +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; +import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.RemoveMountTableEntryRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.RemoveMountTableEntryResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateMountTableEntryRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateMountTableEntryResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.AddMountTableEntryRequestPBImpl; +import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.AddMountTableEntryResponsePBImpl; +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.RemoveMountTableEntryRequestPBImpl; +import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.RemoveMountTableEntryResponsePBImpl; +import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.UpdateMountTableEntryRequestPBImpl; +import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.UpdateMountTableEntryResponsePBImpl; + +import com.google.protobuf.RpcController; +import com.google.protobuf.ServiceException; + +/** + * This class is used on the server side. Calls come across the wire for the for + * protocol {@link RouterAdminProtocolPB}. This class translates the PB data + * types to the native data types used inside the HDFS Router as specified in + * the generic RouterAdminProtocol. + */ +@InterfaceAudience.Private +@InterfaceStability.Stable +public class RouterAdminProtocolServerSideTranslatorPB implements + RouterAdminProtocolPB { + + private final RouterAdminServer server; + + /** + * Constructor. + * @param server The NN server. + * @throws IOException + */ + public RouterAdminProtocolServerSideTranslatorPB(RouterAdminServer server) + throws IOException { + this.server = server; + } + + @Override + public AddMountTableEntryResponseProto addMountTableEntry( + RpcController controller, AddMountTableEntryRequestProto request) + throws ServiceException { + + try { + AddMountTableEntryRequest req = + new AddMountTableEntryRequestPBImpl(request); + AddMountTableEntryResponse response = server.addMountTableEntry(req); + AddMountTableEntryResponsePBImpl responsePB = + (AddMountTableEntryResponsePBImpl)response; + return responsePB.getProto(); + } catch (IOException e) { + throw new ServiceException(e); + } + } + + /** + * Remove an entry from the mount table. + */ + @Override + public RemoveMountTableEntryResponseProto removeMountTableEntry( + RpcController controller, RemoveMountTableEntryRequestProto request) + throws ServiceException { + try { + RemoveMountTableEntryRequest req = + new RemoveMountTableEntryRequestPBImpl(request); + RemoveMountTableEntryResponse response = + server.removeMountTableEntry(req); + RemoveMountTableEntryResponsePBImpl responsePB = + (RemoveMountTableEntryResponsePBImpl)response; + return responsePB.getProto(); + } catch (IOException e) { + throw new ServiceException(e); + } + } + + /** + * Get matching mount table entries. + */ + @Override + public GetMountTableEntriesResponseProto getMountTableEntries( + RpcController controller, GetMountTableEntriesRequestProto request) + throws ServiceException { + try { + GetMountTableEntriesRequest req = + new GetMountTableEntriesRequestPBImpl(request); + GetMountTableEntriesResponse response = server.getMountTableEntries(req); + GetMountTableEntriesResponsePBImpl responsePB = + (GetMountTableEntriesResponsePBImpl)response; + return responsePB.getProto(); + } catch (IOException e) { + throw new ServiceException(e); + } + } + + /** + * Update a single mount table entry. + */ + @Override + public UpdateMountTableEntryResponseProto updateMountTableEntry( + RpcController controller, UpdateMountTableEntryRequestProto request) + throws ServiceException { + try { + UpdateMountTableEntryRequest req = + new UpdateMountTableEntryRequestPBImpl(request); + UpdateMountTableEntryResponse response = + server.updateMountTableEntry(req); + UpdateMountTableEntryResponsePBImpl responsePB = + (UpdateMountTableEntryResponsePBImpl)response; + return responsePB.getProto(); + } catch (IOException e) { + throw new ServiceException(e); + } + } +} diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/RouterAdminProtocolTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/RouterAdminProtocolTranslatorPB.java new file mode 100644 index 00000000000..43663ac5502 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/RouterAdminProtocolTranslatorPB.java @@ -0,0 +1,150 @@ +/** + * 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.protocolPB; + +import java.io.Closeable; +import java.io.IOException; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.classification.InterfaceStability; +import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.AddMountTableEntryRequestProto; +import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.AddMountTableEntryResponseProto; +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.RemoveMountTableEntryRequestProto; +import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.RemoveMountTableEntryResponseProto; +import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.UpdateMountTableEntryRequestProto; +import org.apache.hadoop.hdfs.federation.protocol.proto.HdfsServerFederationProtos.UpdateMountTableEntryResponseProto; +import org.apache.hadoop.hdfs.server.federation.resolver.MountTableManager; +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; +import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.RemoveMountTableEntryRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.RemoveMountTableEntryResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateMountTableEntryRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateMountTableEntryResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.AddMountTableEntryRequestPBImpl; +import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.AddMountTableEntryResponsePBImpl; +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.RemoveMountTableEntryRequestPBImpl; +import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.RemoveMountTableEntryResponsePBImpl; +import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.UpdateMountTableEntryRequestPBImpl; +import org.apache.hadoop.hdfs.server.federation.store.protocol.impl.pb.UpdateMountTableEntryResponsePBImpl; +import org.apache.hadoop.ipc.ProtobufHelper; +import org.apache.hadoop.ipc.ProtocolMetaInterface; +import org.apache.hadoop.ipc.ProtocolTranslator; +import org.apache.hadoop.ipc.RPC; +import org.apache.hadoop.ipc.RpcClientUtil; + +import com.google.protobuf.ServiceException; + +/** + * This class forwards NN's ClientProtocol calls as RPC calls to the NN server + * while translating from the parameter types used in ClientProtocol to the + * new PB types. + */ +@InterfaceAudience.Private +@InterfaceStability.Stable +public class RouterAdminProtocolTranslatorPB + implements ProtocolMetaInterface, MountTableManager, + Closeable, ProtocolTranslator { + final private RouterAdminProtocolPB rpcProxy; + + public RouterAdminProtocolTranslatorPB(RouterAdminProtocolPB proxy) { + rpcProxy = proxy; + } + + @Override + public void close() { + RPC.stopProxy(rpcProxy); + } + + @Override + public Object getUnderlyingProxyObject() { + return rpcProxy; + } + + @Override + public boolean isMethodSupported(String methodName) throws IOException { + return RpcClientUtil.isMethodSupported(rpcProxy, + RouterAdminProtocolPB.class, RPC.RpcKind.RPC_PROTOCOL_BUFFER, + RPC.getProtocolVersion(RouterAdminProtocolPB.class), methodName); + } + + @Override + public AddMountTableEntryResponse addMountTableEntry( + AddMountTableEntryRequest request) throws IOException { + AddMountTableEntryRequestPBImpl requestPB = + (AddMountTableEntryRequestPBImpl)request; + AddMountTableEntryRequestProto proto = requestPB.getProto(); + try { + AddMountTableEntryResponseProto response = + rpcProxy.addMountTableEntry(null, proto); + return new AddMountTableEntryResponsePBImpl(response); + } catch (ServiceException e) { + throw new IOException(ProtobufHelper.getRemoteException(e).getMessage()); + } + } + + @Override + public UpdateMountTableEntryResponse updateMountTableEntry( + UpdateMountTableEntryRequest request) throws IOException { + UpdateMountTableEntryRequestPBImpl requestPB = + (UpdateMountTableEntryRequestPBImpl)request; + UpdateMountTableEntryRequestProto proto = requestPB.getProto(); + try { + UpdateMountTableEntryResponseProto response = + rpcProxy.updateMountTableEntry(null, proto); + return new UpdateMountTableEntryResponsePBImpl(response); + } catch (ServiceException e) { + throw new IOException(ProtobufHelper.getRemoteException(e).getMessage()); + } + } + + @Override + public RemoveMountTableEntryResponse removeMountTableEntry( + RemoveMountTableEntryRequest request) throws IOException { + RemoveMountTableEntryRequestPBImpl requestPB = + (RemoveMountTableEntryRequestPBImpl)request; + RemoveMountTableEntryRequestProto proto = requestPB.getProto(); + try { + RemoveMountTableEntryResponseProto responseProto = + rpcProxy.removeMountTableEntry(null, proto); + return new RemoveMountTableEntryResponsePBImpl(responseProto); + } catch (ServiceException e) { + throw new IOException(ProtobufHelper.getRemoteException(e).getMessage()); + } + } + + @Override + public GetMountTableEntriesResponse getMountTableEntries( + GetMountTableEntriesRequest request) throws IOException { + GetMountTableEntriesRequestPBImpl requestPB = + (GetMountTableEntriesRequestPBImpl)request; + GetMountTableEntriesRequestProto proto = requestPB.getProto(); + try { + GetMountTableEntriesResponseProto response = + rpcProxy.getMountTableEntries(null, proto); + return new GetMountTableEntriesResponsePBImpl(response); + } catch (ServiceException e) { + throw new IOException(ProtobufHelper.getRemoteException(e).getMessage()); + } + } +} diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MembershipNamenodeResolver.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MembershipNamenodeResolver.java index b0ced24a8ae..d974c78e64c 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MembershipNamenodeResolver.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MembershipNamenodeResolver.java @@ -61,7 +61,7 @@ public class MembershipNamenodeResolver /** Reference to the State Store. */ private final StateStoreService stateStore; /** Membership State Store interface. */ - private final MembershipStore membershipInterface; + private MembershipStore membershipInterface; /** Parent router ID. */ private String routerId; @@ -82,25 +82,27 @@ public class MembershipNamenodeResolver if (this.stateStore != null) { // Request cache updates from the state store this.stateStore.registerCacheExternal(this); + } + } - // Initialize the interface to get the membership + private synchronized MembershipStore getMembershipStore() throws IOException { + if (this.membershipInterface == null) { this.membershipInterface = this.stateStore.getRegisteredRecordStore( MembershipStore.class); - } else { - this.membershipInterface = null; - } - - if (this.membershipInterface == null) { - throw new IOException("State Store does not have an interface for " + - MembershipStore.class.getSimpleName()); + if (this.membershipInterface == null) { + throw new IOException("State Store does not have an interface for " + + MembershipStore.class.getSimpleName()); + } } + return this.membershipInterface; } @Override public boolean loadCache(boolean force) { // Our cache depends on the store, update it first try { - this.membershipInterface.loadCache(force); + MembershipStore membership = getMembershipStore(); + membership.loadCache(force); } catch (IOException e) { LOG.error("Cannot update membership from the State Store", e); } @@ -126,8 +128,9 @@ public class MembershipNamenodeResolver GetNamenodeRegistrationsRequest request = GetNamenodeRegistrationsRequest.newInstance(partial); + MembershipStore membership = getMembershipStore(); GetNamenodeRegistrationsResponse response = - this.membershipInterface.getNamenodeRegistrations(request); + membership.getNamenodeRegistrations(request); List records = response.getNamenodeMemberships(); if (records != null && records.size() == 1) { @@ -135,7 +138,7 @@ public class MembershipNamenodeResolver UpdateNamenodeRegistrationRequest updateRequest = UpdateNamenodeRegistrationRequest.newInstance( record.getNameserviceId(), record.getNamenodeId(), ACTIVE); - this.membershipInterface.updateNamenodeRegistration(updateRequest); + membership.updateNamenodeRegistration(updateRequest); } } catch (StateStoreUnavailableException e) { LOG.error("Cannot update {} as active, State Store unavailable", address); @@ -226,14 +229,14 @@ public class MembershipNamenodeResolver NamenodeHeartbeatRequest request = NamenodeHeartbeatRequest.newInstance(); request.setNamenodeMembership(record); - return this.membershipInterface.namenodeHeartbeat(request).getResult(); + return getMembershipStore().namenodeHeartbeat(request).getResult(); } @Override public Set getNamespaces() throws IOException { GetNamespaceInfoRequest request = GetNamespaceInfoRequest.newInstance(); GetNamespaceInfoResponse response = - this.membershipInterface.getNamespaceInfo(request); + getMembershipStore().getNamespaceInfo(request); return response.getNamespaceInfo(); } @@ -259,8 +262,9 @@ public class MembershipNamenodeResolver // Retrieve a list of all registrations that match this query. // This may include all NN records for a namespace/blockpool, including // duplicate records for the same NN from different routers. + MembershipStore membershipStore = getMembershipStore(); GetNamenodeRegistrationsResponse response = - this.membershipInterface.getNamenodeRegistrations(request); + membershipStore.getNamenodeRegistrations(request); List memberships = response.getNamenodeMemberships(); if (!addExpired || !addUnavailable) { diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/Router.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/Router.java index 213a58fc110..fcbd2eb3eba 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/Router.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/Router.java @@ -81,6 +81,10 @@ public class Router extends CompositeService { private RouterRpcServer rpcServer; private InetSocketAddress rpcAddress; + /** RPC interface for the admin. */ + private RouterAdminServer adminServer; + private InetSocketAddress adminAddress; + /** Interface with the State Store. */ private StateStoreService stateStore; @@ -116,6 +120,14 @@ public class Router extends CompositeService { protected void serviceInit(Configuration configuration) throws Exception { this.conf = configuration; + if (conf.getBoolean( + DFSConfigKeys.DFS_ROUTER_STORE_ENABLE, + DFSConfigKeys.DFS_ROUTER_STORE_ENABLE_DEFAULT)) { + // Service that maintains the State Store connection + this.stateStore = new StateStoreService(); + addService(this.stateStore); + } + // Resolver to track active NNs this.namenodeResolver = newActiveNamenodeResolver( this.conf, this.stateStore); @@ -138,6 +150,14 @@ public class Router extends CompositeService { this.setRpcServerAddress(rpcServer.getRpcAddress()); } + if (conf.getBoolean( + DFSConfigKeys.DFS_ROUTER_ADMIN_ENABLE, + DFSConfigKeys.DFS_ROUTER_ADMIN_ENABLE_DEFAULT)) { + // Create admin server + this.adminServer = createAdminServer(); + addService(this.adminServer); + } + if (conf.getBoolean( DFSConfigKeys.DFS_ROUTER_HEARTBEAT_ENABLE, DFSConfigKeys.DFS_ROUTER_HEARTBEAT_ENABLE_DEFAULT)) { @@ -263,6 +283,38 @@ public class Router extends CompositeService { return this.rpcAddress; } + ///////////////////////////////////////////////////////// + // Admin server + ///////////////////////////////////////////////////////// + + /** + * Create a new router admin server to handle the router admin interface. + * + * @return RouterAdminServer + * @throws IOException If the admin server was not successfully started. + */ + protected RouterAdminServer createAdminServer() throws IOException { + return new RouterAdminServer(this.conf, this); + } + + /** + * Set the current Admin socket for the router. + * + * @param adminAddress Admin RPC address. + */ + protected void setAdminServerAddress(InetSocketAddress address) { + this.adminAddress = address; + } + + /** + * Get the current Admin socket address for the router. + * + * @return InetSocketAddress Admin address. + */ + public InetSocketAddress getAdminServerAddress() { + return adminAddress; + } + ///////////////////////////////////////////////////////// // Namenode heartbeat monitors ///////////////////////////////////////////////////////// diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterAdminServer.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterAdminServer.java new file mode 100644 index 00000000000..7687216a98a --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterAdminServer.java @@ -0,0 +1,183 @@ +/** + * 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.router; + +import java.io.IOException; +import java.net.InetSocketAddress; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hdfs.DFSConfigKeys; +import org.apache.hadoop.hdfs.protocol.proto.RouterProtocolProtos.RouterAdminProtocolService; +import org.apache.hadoop.hdfs.protocolPB.RouterAdminProtocolPB; +import org.apache.hadoop.hdfs.protocolPB.RouterAdminProtocolServerSideTranslatorPB; +import org.apache.hadoop.hdfs.server.federation.resolver.MountTableManager; +import org.apache.hadoop.hdfs.server.federation.store.MountTableStore; +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; +import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.RemoveMountTableEntryRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.RemoveMountTableEntryResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateMountTableEntryRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateMountTableEntryResponse; +import org.apache.hadoop.ipc.ProtobufRpcEngine; +import org.apache.hadoop.ipc.RPC; +import org.apache.hadoop.ipc.RPC.Server; +import org.apache.hadoop.service.AbstractService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.annotations.VisibleForTesting; +import com.google.protobuf.BlockingService; + +/** + * This class is responsible for handling all of the Admin calls to the HDFS + * router. It is created, started, and stopped by {@link Router}. + */ +public class RouterAdminServer extends AbstractService + implements MountTableManager { + + private static final Logger LOG = + LoggerFactory.getLogger(RouterAdminServer.class); + + private Configuration conf; + + private final Router router; + + private MountTableStore mountTableStore; + + /** The Admin server that listens to requests from clients. */ + private final Server adminServer; + private final InetSocketAddress adminAddress; + + public RouterAdminServer(Configuration conf, Router router) + throws IOException { + super(RouterAdminServer.class.getName()); + + this.conf = conf; + this.router = router; + + int handlerCount = this.conf.getInt( + DFSConfigKeys.DFS_ROUTER_ADMIN_HANDLER_COUNT_KEY, + DFSConfigKeys.DFS_ROUTER_ADMIN_HANDLER_COUNT_DEFAULT); + + RPC.setProtocolEngine(this.conf, RouterAdminProtocolPB.class, + ProtobufRpcEngine.class); + + RouterAdminProtocolServerSideTranslatorPB routerAdminProtocolTranslator = + new RouterAdminProtocolServerSideTranslatorPB(this); + BlockingService clientNNPbService = RouterAdminProtocolService. + newReflectiveBlockingService(routerAdminProtocolTranslator); + + InetSocketAddress confRpcAddress = conf.getSocketAddr( + DFSConfigKeys.DFS_ROUTER_ADMIN_BIND_HOST_KEY, + DFSConfigKeys.DFS_ROUTER_ADMIN_ADDRESS_KEY, + DFSConfigKeys.DFS_ROUTER_ADMIN_ADDRESS_DEFAULT, + DFSConfigKeys.DFS_ROUTER_ADMIN_PORT_DEFAULT); + + String bindHost = conf.get( + DFSConfigKeys.DFS_ROUTER_ADMIN_BIND_HOST_KEY, + confRpcAddress.getHostName()); + LOG.info("Admin server binding to {}:{}", + bindHost, confRpcAddress.getPort()); + + this.adminServer = new RPC.Builder(this.conf) + .setProtocol(RouterAdminProtocolPB.class) + .setInstance(clientNNPbService) + .setBindAddress(bindHost) + .setPort(confRpcAddress.getPort()) + .setNumHandlers(handlerCount) + .setVerbose(false) + .build(); + + // The RPC-server port can be ephemeral... ensure we have the correct info + InetSocketAddress listenAddress = this.adminServer.getListenerAddress(); + this.adminAddress = new InetSocketAddress( + confRpcAddress.getHostName(), listenAddress.getPort()); + router.setAdminServerAddress(this.adminAddress); + } + + /** Allow access to the client RPC server for testing. */ + @VisibleForTesting + Server getAdminServer() { + return this.adminServer; + } + + private MountTableStore getMountTableStore() throws IOException { + if (this.mountTableStore == null) { + this.mountTableStore = router.getStateStore().getRegisteredRecordStore( + MountTableStore.class); + if (this.mountTableStore == null) { + throw new IOException("Mount table state store is not available."); + } + } + return this.mountTableStore; + } + + /** + * Get the RPC address of the admin service. + * @return Administration service RPC address. + */ + public InetSocketAddress getRpcAddress() { + return this.adminAddress; + } + + @Override + protected void serviceInit(Configuration configuration) throws Exception { + this.conf = configuration; + super.serviceInit(conf); + } + + @Override + protected void serviceStart() throws Exception { + this.adminServer.start(); + super.serviceStart(); + } + + @Override + protected void serviceStop() throws Exception { + if (this.adminServer != null) { + this.adminServer.stop(); + } + super.serviceStop(); + } + + @Override + public AddMountTableEntryResponse addMountTableEntry( + AddMountTableEntryRequest request) throws IOException { + return getMountTableStore().addMountTableEntry(request); + } + + @Override + public UpdateMountTableEntryResponse updateMountTableEntry( + UpdateMountTableEntryRequest request) throws IOException { + return getMountTableStore().updateMountTableEntry(request); + } + + @Override + public RemoveMountTableEntryResponse removeMountTableEntry( + RemoveMountTableEntryRequest request) throws IOException { + return getMountTableStore().removeMountTableEntry(request); + } + + @Override + public GetMountTableEntriesResponse getMountTableEntries( + GetMountTableEntriesRequest request) throws IOException { + return getMountTableStore().getMountTableEntries(request); + } +} diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterClient.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterClient.java new file mode 100644 index 00000000000..1f76b980108 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterClient.java @@ -0,0 +1,76 @@ +/** + * 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.router; + +import java.io.Closeable; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.hadoop.classification.InterfaceAudience.Private; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hdfs.protocolPB.RouterAdminProtocolPB; +import org.apache.hadoop.hdfs.protocolPB.RouterAdminProtocolTranslatorPB; +import org.apache.hadoop.hdfs.server.federation.resolver.MountTableManager; +import org.apache.hadoop.ipc.ProtobufRpcEngine; +import org.apache.hadoop.ipc.RPC; +import org.apache.hadoop.net.NetUtils; +import org.apache.hadoop.security.UserGroupInformation; + +/** + * Client to connect to the {@link Router} via the admin protocol. + */ +@Private +public class RouterClient implements Closeable { + + private final RouterAdminProtocolTranslatorPB proxy; + private final UserGroupInformation ugi; + + private static RouterAdminProtocolTranslatorPB createRouterProxy( + InetSocketAddress address, Configuration conf, UserGroupInformation ugi) + throws IOException { + + RPC.setProtocolEngine( + conf, RouterAdminProtocolPB.class, ProtobufRpcEngine.class); + + AtomicBoolean fallbackToSimpleAuth = new AtomicBoolean(false); + final long version = RPC.getProtocolVersion(RouterAdminProtocolPB.class); + RouterAdminProtocolPB proxy = RPC.getProtocolProxy( + RouterAdminProtocolPB.class, version, address, ugi, conf, + NetUtils.getDefaultSocketFactory(conf), + RPC.getRpcTimeout(conf), null, + fallbackToSimpleAuth).getProxy(); + + return new RouterAdminProtocolTranslatorPB(proxy); + } + + public RouterClient(InetSocketAddress address, Configuration conf) + throws IOException { + this.ugi = UserGroupInformation.getCurrentUser(); + this.proxy = createRouterProxy(address, conf, ugi); + } + + public MountTableManager getMountTableManager() { + return proxy; + } + + @Override + public synchronized void close() throws IOException { + RPC.stopProxy(proxy); + } +} \ No newline at end of file diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/federation/RouterAdmin.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/federation/RouterAdmin.java new file mode 100644 index 00000000000..07864197f61 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/federation/RouterAdmin.java @@ -0,0 +1,341 @@ +/** + * 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.tools.federation; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import org.apache.hadoop.classification.InterfaceAudience.Private; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.conf.Configured; +import org.apache.hadoop.hdfs.DFSConfigKeys; +import org.apache.hadoop.hdfs.HdfsConfiguration; +import org.apache.hadoop.hdfs.server.federation.resolver.MountTableManager; +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.router.RouterClient; +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; +import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.RemoveMountTableEntryRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.RemoveMountTableEntryResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateMountTableEntryRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateMountTableEntryResponse; +import org.apache.hadoop.hdfs.server.federation.store.records.MountTable; +import org.apache.hadoop.ipc.RPC; +import org.apache.hadoop.ipc.RemoteException; +import org.apache.hadoop.net.NetUtils; +import org.apache.hadoop.util.Tool; +import org.apache.hadoop.util.ToolRunner; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * This class provides some Federation administrative access shell commands. + */ +@Private +public class RouterAdmin extends Configured implements Tool { + + private static final Logger LOG = LoggerFactory.getLogger(RouterAdmin.class); + + private RouterClient client; + + public static void main(String[] argv) throws Exception { + Configuration conf = new HdfsConfiguration(); + RouterAdmin admin = new RouterAdmin(conf); + + int res = ToolRunner.run(admin, argv); + System.exit(res); + } + + public RouterAdmin(Configuration conf) { + super(conf); + } + + /** + * Print the usage message. + */ + public void printUsage() { + String usage = "Federation Admin Tools:\n" + + "\t[-add " + + "[-readonly] [-order HASH|LOCAL|RANDOM|HASH_ALL]]\n" + + "\t[-rm ]\n" + + "\t[-ls ]\n"; + System.out.println(usage); + } + + @Override + public int run(String[] argv) throws Exception { + if (argv.length < 1) { + System.err.println("Not enough parameters specificed"); + printUsage(); + return -1; + } + + int exitCode = -1; + int i = 0; + String cmd = argv[i++]; + + // Verify that we have enough command line parameters + if ("-add".equals(cmd)) { + if (argv.length < 4) { + System.err.println("Not enough parameters specificed for cmd " + cmd); + printUsage(); + return exitCode; + } + } else if ("-rm".equalsIgnoreCase(cmd)) { + if (argv.length < 2) { + System.err.println("Not enough parameters specificed for cmd " + cmd); + printUsage(); + return exitCode; + } + } + + // Initialize RouterClient + try { + String address = getConf().getTrimmed( + DFSConfigKeys.DFS_ROUTER_ADMIN_ADDRESS_KEY, + DFSConfigKeys.DFS_ROUTER_ADMIN_ADDRESS_DEFAULT); + InetSocketAddress routerSocket = NetUtils.createSocketAddr(address); + client = new RouterClient(routerSocket, getConf()); + } catch (RPC.VersionMismatch v) { + System.err.println( + "Version mismatch between client and server... command aborted"); + return exitCode; + } catch (IOException e) { + System.err.println("Bad connection to Router... command aborted"); + return exitCode; + } + + Exception debugException = null; + exitCode = 0; + try { + if ("-add".equals(cmd)) { + if (addMount(argv, i)) { + System.err.println("Successfuly added mount point " + argv[i]); + } + } else if ("-rm".equals(cmd)) { + if (removeMount(argv[i])) { + System.err.println("Successfully removed mount point " + argv[i]); + } + } else if ("-ls".equals(cmd)) { + if (argv.length > 1) { + listMounts(argv[i]); + } else { + listMounts("/"); + } + } else { + printUsage(); + return exitCode; + } + } catch (IllegalArgumentException arge) { + debugException = arge; + exitCode = -1; + System.err.println(cmd.substring(1) + ": " + arge.getLocalizedMessage()); + printUsage(); + } catch (RemoteException e) { + // This is a error returned by the server. + // Print out the first line of the error message, ignore the stack trace. + exitCode = -1; + debugException = e; + try { + String[] content; + content = e.getLocalizedMessage().split("\n"); + System.err.println(cmd.substring(1) + ": " + content[0]); + e.printStackTrace(); + } catch (Exception ex) { + System.err.println(cmd.substring(1) + ": " + ex.getLocalizedMessage()); + e.printStackTrace(); + debugException = ex; + } + } catch (Exception e) { + exitCode = -1; + debugException = e; + System.err.println(cmd.substring(1) + ": " + e.getLocalizedMessage()); + e.printStackTrace(); + } + if (debugException != null) { + LOG.debug("Exception encountered", debugException); + } + return exitCode; + } + + /** + * Add a mount table entry or update if it exists. + * + * @param parameters Parameters for the mount point. + * @param i Index in the parameters. + */ + public boolean addMount(String[] parameters, int i) throws IOException { + // Mandatory parameters + String mount = parameters[i++]; + String[] nss = parameters[i++].split(","); + String dest = parameters[i++]; + + // Optional parameters + boolean readOnly = false; + DestinationOrder order = DestinationOrder.HASH; + while (i < parameters.length) { + if (parameters[i].equals("-readonly")) { + readOnly = true; + } else if (parameters[i].equals("-order")) { + i++; + try { + order = DestinationOrder.valueOf(parameters[i]); + } catch(Exception e) { + System.err.println("Cannot parse order: " + parameters[i]); + } + } + i++; + } + + return addMount(mount, nss, dest, readOnly, order); + } + + /** + * Add a mount table entry or update if it exists. + * + * @param mount Mount point. + * @param nss Namespaces where this is mounted to. + * @param dest Destination path. + * @param readonly If the mount point is read only. + * @param order Order of the destination locations. + * @return If the mount point was added. + * @throws IOException Error adding the mount point. + */ + public boolean addMount(String mount, String[] nss, String dest, + boolean readonly, DestinationOrder order) throws IOException { + // Get the existing entry + MountTableManager mountTable = client.getMountTableManager(); + GetMountTableEntriesRequest getRequest = + GetMountTableEntriesRequest.newInstance(mount); + GetMountTableEntriesResponse getResponse = + mountTable.getMountTableEntries(getRequest); + List results = getResponse.getEntries(); + MountTable existingEntry = null; + for (MountTable result : results) { + if (mount.equals(result.getSourcePath())) { + existingEntry = result; + } + } + + if (existingEntry == null) { + // Create and add the entry if it doesn't exist + Map destMap = new LinkedHashMap<>(); + for (String ns : nss) { + destMap.put(ns, dest); + } + MountTable newEntry = MountTable.newInstance(mount, destMap); + if (readonly) { + newEntry.setReadOnly(true); + } + if (order != null) { + newEntry.setDestOrder(order); + } + AddMountTableEntryRequest request = + AddMountTableEntryRequest.newInstance(newEntry); + AddMountTableEntryResponse addResponse = + mountTable.addMountTableEntry(request); + boolean added = addResponse.getStatus(); + if (!added) { + System.err.println("Cannot add mount point " + mount); + } + return added; + } else { + // Update the existing entry if it exists + for (String nsId : nss) { + if (!existingEntry.addDestination(nsId, dest)) { + System.err.println("Cannot add destination at " + nsId + " " + dest); + } + } + if (readonly) { + existingEntry.setReadOnly(true); + } + if (order != null) { + existingEntry.setDestOrder(order); + } + UpdateMountTableEntryRequest updateRequest = + UpdateMountTableEntryRequest.newInstance(existingEntry); + UpdateMountTableEntryResponse updateResponse = + mountTable.updateMountTableEntry(updateRequest); + boolean updated = updateResponse.getStatus(); + if (!updated) { + System.err.println("Cannot update mount point " + mount); + } + return updated; + } + } + + /** + * Remove mount point. + * + * @param path Path to remove. + * @throws IOException If it cannot be removed. + */ + public boolean removeMount(String path) throws IOException { + MountTableManager mountTable = client.getMountTableManager(); + RemoveMountTableEntryRequest request = + RemoveMountTableEntryRequest.newInstance(path); + RemoveMountTableEntryResponse response = + mountTable.removeMountTableEntry(request); + boolean removed = response.getStatus(); + if (!removed) { + System.out.println("Cannot remove mount point " + path); + } + return removed; + } + + /** + * List mount points. + * + * @param path Path to list. + * @throws IOException If it cannot be listed. + */ + public void listMounts(String path) throws IOException { + MountTableManager mountTable = client.getMountTableManager(); + GetMountTableEntriesRequest request = + GetMountTableEntriesRequest.newInstance(path); + GetMountTableEntriesResponse response = + mountTable.getMountTableEntries(request); + List entries = response.getEntries(); + printMounts(entries); + } + + private static void printMounts(List entries) { + System.out.println("Mount Table Entries:"); + System.out.println(String.format( + "%-25s %-25s", + "Source", "Destinations")); + for (MountTable entry : entries) { + StringBuilder destBuilder = new StringBuilder(); + for (RemoteLocation location : entry.getDestinations()) { + if (destBuilder.length() > 0) { + destBuilder.append(","); + } + destBuilder.append(String.format("%s->%s", location.getNameserviceId(), + location.getDest())); + } + System.out.println(String.format("%-25s %-25s", entry.getSourcePath(), + destBuilder.toString())); + } + } +} \ No newline at end of file diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/federation/package-info.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/federation/package-info.java new file mode 100644 index 00000000000..466c3d3c91c --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/federation/package-info.java @@ -0,0 +1,28 @@ +/** + * 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. + */ + +/** + * It includes the tools to manage the Router-based federation. Includes the + * utilities to add and remove mount table entries. + */ +@InterfaceAudience.Public +@InterfaceStability.Evolving +package org.apache.hadoop.hdfs.tools.federation; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.classification.InterfaceStability; \ No newline at end of file diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/proto/RouterProtocol.proto b/hadoop-hdfs-project/hadoop-hdfs/src/main/proto/RouterProtocol.proto new file mode 100644 index 00000000000..3f4304024ba --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/proto/RouterProtocol.proto @@ -0,0 +1,47 @@ +/** + * 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. + */ + +option java_package = "org.apache.hadoop.hdfs.protocol.proto"; +option java_outer_classname = "RouterProtocolProtos"; +option java_generic_services = true; +option java_generate_equals_and_hash = true; +package hadoop.hdfs.router; + +import "FederationProtocol.proto"; + +service RouterAdminProtocolService { + /** + * Add a mount table entry. + */ + rpc addMountTableEntry(AddMountTableEntryRequestProto) returns(AddMountTableEntryResponseProto); + + /** + * Update an existing mount table entry without copying files. + */ + rpc updateMountTableEntry(UpdateMountTableEntryRequestProto) returns(UpdateMountTableEntryResponseProto); + + /** + * Remove a mount table entry. + */ + rpc removeMountTableEntry(RemoveMountTableEntryRequestProto) returns(RemoveMountTableEntryResponseProto); + + /** + * Get matching mount entries + */ + rpc getMountTableEntries(GetMountTableEntriesRequestProto) returns(GetMountTableEntriesResponseProto); +} \ No newline at end of file diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml index cc4a2bfdf16..7f421413767 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml @@ -4745,6 +4745,44 @@ + + dfs.federation.router.admin.enable + true + + If the RPC admin service to handle client requests in the router is + enabled. + + + + + dfs.federation.router.admin-address + 0.0.0.0:8111 + + RPC address that handles the admin requests. + The value of this property will take the form of router-host1:rpc-port. + + + + + dfs.federation.router.admin-bind-host + + + The actual address the RPC admin server will bind to. If this optional + address is set, it overrides only the hostname portion of + dfs.federation.router.admin-address. This is useful for making the name + node listen on all interfaces by setting it to 0.0.0.0. + + + + + dfs.federation.router.admin.handler.count + 1 + + The number of server threads for the router to handle RPC requests from + admin. + + + dfs.federation.router.file.resolver.client.class org.apache.hadoop.hdfs.server.federation.MockResolver @@ -4761,6 +4799,14 @@ + + dfs.federation.router.store.enable + true + + If the Router connects to the State Store. + + + dfs.federation.router.store.serializer org.apache.hadoop.hdfs.server.federation.store.driver.impl.StateStoreSerializerPBImpl diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/RouterConfigBuilder.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/RouterConfigBuilder.java index 21555c55a62..cac5e6b6078 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/RouterConfigBuilder.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/RouterConfigBuilder.java @@ -28,8 +28,10 @@ public class RouterConfigBuilder { private Configuration conf; private boolean enableRpcServer = false; + private boolean enableAdminServer = false; private boolean enableHeartbeat = false; private boolean enableLocalHeartbeat = false; + private boolean enableStateStore = false; public RouterConfigBuilder(Configuration configuration) { this.conf = configuration; @@ -41,8 +43,10 @@ public class RouterConfigBuilder { public RouterConfigBuilder all() { this.enableRpcServer = true; + this.enableAdminServer = true; this.enableHeartbeat = true; this.enableLocalHeartbeat = true; + this.enableStateStore = true; return this; } @@ -56,21 +60,43 @@ public class RouterConfigBuilder { return this; } + public RouterConfigBuilder admin(boolean enable) { + this.enableAdminServer = enable; + return this; + } + public RouterConfigBuilder heartbeat(boolean enable) { this.enableHeartbeat = enable; return this; } + public RouterConfigBuilder stateStore(boolean enable) { + this.enableStateStore = enable; + return this; + } + public RouterConfigBuilder rpc() { return this.rpc(true); } + public RouterConfigBuilder admin() { + return this.admin(true); + } + public RouterConfigBuilder heartbeat() { return this.heartbeat(true); } + public RouterConfigBuilder stateStore() { + return this.stateStore(true); + } + public Configuration build() { + conf.setBoolean(DFSConfigKeys.DFS_ROUTER_STORE_ENABLE, + this.enableStateStore); conf.setBoolean(DFSConfigKeys.DFS_ROUTER_RPC_ENABLE, this.enableRpcServer); + conf.setBoolean(DFSConfigKeys.DFS_ROUTER_ADMIN_ENABLE, + this.enableAdminServer); conf.setBoolean(DFSConfigKeys.DFS_ROUTER_HEARTBEAT_ENABLE, this.enableHeartbeat); conf.setBoolean(DFSConfigKeys.DFS_ROUTER_MONITOR_LOCAL_NAMENODE, diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/RouterDFSCluster.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/RouterDFSCluster.java index 0830c197bfd..1ee49d566a6 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/RouterDFSCluster.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/RouterDFSCluster.java @@ -25,8 +25,12 @@ import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_RPC_ADDRESS_KEY; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_RPC_BIND_HOST_KEY; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMESERVICES; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMESERVICE_ID; +import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_ROUTER_ADMIN_ADDRESS_KEY; +import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_ROUTER_ADMIN_BIND_HOST_KEY; +import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_ROUTER_CACHE_TIME_TO_LIVE_MS; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_ROUTER_DEFAULT_NAMESERVICE; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_ROUTER_HANDLER_COUNT_KEY; +import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_ROUTER_HEARTBEAT_INTERVAL_MS; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_ROUTER_RPC_ADDRESS_KEY; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_ROUTER_RPC_BIND_HOST_KEY; import static org.apache.hadoop.hdfs.DFSConfigKeys.FEDERATION_FILE_RESOLVER_CLIENT_CLASS; @@ -46,6 +50,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Map.Entry; import java.util.Random; +import java.util.concurrent.TimeUnit; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileContext; @@ -67,6 +72,7 @@ import org.apache.hadoop.hdfs.server.federation.resolver.FederationNamenodeServi import org.apache.hadoop.hdfs.server.federation.resolver.FileSubclusterResolver; import org.apache.hadoop.hdfs.server.federation.resolver.NamenodeStatusReport; import org.apache.hadoop.hdfs.server.federation.router.Router; +import org.apache.hadoop.hdfs.server.federation.router.RouterClient; import org.apache.hadoop.hdfs.server.namenode.FSImage; import org.apache.hadoop.hdfs.server.namenode.NameNode; import org.apache.hadoop.hdfs.server.protocol.NamespaceInfo; @@ -101,6 +107,15 @@ public class RouterDFSCluster { /** Mini cluster. */ private MiniDFSCluster cluster; + protected static final long DEFAULT_HEARTBEAT_INTERVAL_MS = + TimeUnit.SECONDS.toMillis(5); + protected static final long DEFAULT_CACHE_INTERVAL_MS = + TimeUnit.SECONDS.toMillis(5); + /** Heartbeat interval in milliseconds. */ + private long heartbeatInterval; + /** Cache flush interval in milliseconds. */ + private long cacheFlushInterval; + /** Router configuration overrides. */ private Configuration routerOverrides; /** Namenode configuration overrides. */ @@ -118,6 +133,7 @@ public class RouterDFSCluster { private int rpcPort; private DFSClient client; private Configuration conf; + private RouterClient adminClient; private URI fileSystemUri; public RouterContext(Configuration conf, String nsId, String nnId) @@ -183,6 +199,15 @@ public class RouterDFSCluster { }); } + public RouterClient getAdminClient() throws IOException { + if (adminClient == null) { + InetSocketAddress routerSocket = router.getAdminServerAddress(); + LOG.info("Connecting to router admin at {}", routerSocket); + adminClient = new RouterClient(routerSocket, conf); + } + return adminClient; + } + public DFSClient getClient() throws IOException, URISyntaxException { if (client == null) { LOG.info("Connecting to router at {}", fileSystemUri); @@ -304,13 +329,22 @@ public class RouterDFSCluster { } } - public RouterDFSCluster(boolean ha, int numNameservices, int numNamenodes) { + public RouterDFSCluster(boolean ha, int numNameservices, int numNamenodes, + long heartbeatInterval, long cacheFlushInterval) { this.highAvailability = ha; + this.heartbeatInterval = heartbeatInterval; + this.cacheFlushInterval = cacheFlushInterval; configureNameservices(numNameservices, numNamenodes); } public RouterDFSCluster(boolean ha, int numNameservices) { - this(ha, numNameservices, 2); + this(ha, numNameservices, 2, + DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_CACHE_INTERVAL_MS); + } + + public RouterDFSCluster(boolean ha, int numNameservices, int numNamnodes) { + this(ha, numNameservices, numNamnodes, + DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_CACHE_INTERVAL_MS); } /** @@ -404,7 +438,12 @@ public class RouterDFSCluster { conf.set(DFS_ROUTER_RPC_ADDRESS_KEY, "127.0.0.1:0"); conf.set(DFS_ROUTER_RPC_BIND_HOST_KEY, "0.0.0.0"); + conf.set(DFS_ROUTER_ADMIN_ADDRESS_KEY, "127.0.0.1:0"); + conf.set(DFS_ROUTER_ADMIN_BIND_HOST_KEY, "0.0.0.0"); + conf.set(DFS_ROUTER_DEFAULT_NAMESERVICE, nameservices.get(0)); + conf.setLong(DFS_ROUTER_HEARTBEAT_INTERVAL_MS, heartbeatInterval); + conf.setLong(DFS_ROUTER_CACHE_TIME_TO_LIVE_MS, cacheFlushInterval); // Use mock resolver classes conf.setClass(FEDERATION_NAMENODE_RESOLVER_CLIENT_CLASS, diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/StateStoreDFSCluster.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/StateStoreDFSCluster.java new file mode 100644 index 00000000000..e42ab500d53 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/StateStoreDFSCluster.java @@ -0,0 +1,148 @@ +/** + * 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; + +import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.createMockRegistrationForNamenode; +import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.getStateStoreConfiguration; +import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.synchronizeRecords; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hdfs.DFSConfigKeys; +import org.apache.hadoop.hdfs.server.federation.resolver.ActiveNamenodeResolver; +import org.apache.hadoop.hdfs.server.federation.resolver.FederationNamenodeServiceState; +import org.apache.hadoop.hdfs.server.federation.resolver.FileSubclusterResolver; +import org.apache.hadoop.hdfs.server.federation.resolver.MembershipNamenodeResolver; +import org.apache.hadoop.hdfs.server.federation.resolver.MountTableResolver; +import org.apache.hadoop.hdfs.server.federation.store.StateStoreService; +import org.apache.hadoop.hdfs.server.federation.store.records.MembershipState; +import org.apache.hadoop.hdfs.server.federation.store.records.MountTable; + +/** + * Test utility to mimic a federated HDFS cluster with a router and a state + * store. + */ +public class StateStoreDFSCluster extends RouterDFSCluster { + + private static final Class DEFAULT_FILE_RESOLVER = + MountTableResolver.class; + private static final Class DEFAULT_NAMENODE_RESOLVER = + MembershipNamenodeResolver.class; + + public StateStoreDFSCluster(boolean ha, int numNameservices, int numNamenodes, + long heartbeatInterval, long cacheFlushInterval) + throws IOException, InterruptedException { + this(ha, numNameservices, numNamenodes, heartbeatInterval, + cacheFlushInterval, DEFAULT_FILE_RESOLVER); + } + + public StateStoreDFSCluster(boolean ha, int numNameservices, int numNamenodes, + long heartbeatInterval, long cacheFlushInterval, Class fileResolver) + throws IOException, InterruptedException { + super(ha, numNameservices, numNamenodes, heartbeatInterval, + cacheFlushInterval); + + // Attach state store and resolvers to router + Configuration stateStoreConfig = getStateStoreConfiguration(); + // Use state store backed resolvers + stateStoreConfig.setClass( + DFSConfigKeys.FEDERATION_NAMENODE_RESOLVER_CLIENT_CLASS, + DEFAULT_NAMENODE_RESOLVER, ActiveNamenodeResolver.class); + stateStoreConfig.setClass( + DFSConfigKeys.FEDERATION_FILE_RESOLVER_CLIENT_CLASS, + fileResolver, FileSubclusterResolver.class); + this.addRouterOverrides(stateStoreConfig); + } + + public StateStoreDFSCluster(boolean ha, int numNameservices, + Class fileResolver) throws IOException, InterruptedException { + this(ha, numNameservices, 2, + DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_CACHE_INTERVAL_MS, fileResolver); + } + + public StateStoreDFSCluster(boolean ha, int numNameservices) + throws IOException, InterruptedException { + this(ha, numNameservices, 2, + DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_CACHE_INTERVAL_MS); + } + + public StateStoreDFSCluster(boolean ha, int numNameservices, + int numNamnodes) throws IOException, InterruptedException { + this(ha, numNameservices, numNamnodes, + DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_CACHE_INTERVAL_MS); + } + + ///////////////////////////////////////////////////////////////////////////// + // State Store Test Fixtures + ///////////////////////////////////////////////////////////////////////////// + + /** + * Adds test fixtures for NN registation for each NN nameservice -> NS + * namenode -> NN rpcAddress -> 0.0.0.0:0 webAddress -> 0.0.0.0:0 state -> + * STANDBY safeMode -> false blockPool -> test. + * + * @param stateStore State Store. + * @throws IOException If it cannot register. + */ + public void createTestRegistration(StateStoreService stateStore) + throws IOException { + List entries = new ArrayList(); + for (NamenodeContext nn : this.getNamenodes()) { + MembershipState entry = createMockRegistrationForNamenode( + nn.getNameserviceId(), nn.getNamenodeId(), + FederationNamenodeServiceState.STANDBY); + entries.add(entry); + } + synchronizeRecords( + stateStore, entries, MembershipState.class); + } + + public void createTestMountTable(StateStoreService stateStore) + throws IOException { + List mounts = generateMockMountTable(); + synchronizeRecords(stateStore, mounts, MountTable.class); + stateStore.refreshCaches(); + } + + public List generateMockMountTable() throws IOException { + // create table entries + List entries = new ArrayList<>(); + for (String ns : this.getNameservices()) { + Map destMap = new HashMap<>(); + destMap.put(ns, getNamenodePathForNS(ns)); + + // Direct path + String fedPath = getFederatedPathForNS(ns); + MountTable entry = MountTable.newInstance(fedPath, destMap); + entries.add(entry); + } + + // Root path goes to nameservice 1 + Map destMap = new HashMap<>(); + String ns0 = this.getNameservices().get(0); + destMap.put(ns0, "/"); + MountTable entry = MountTable.newInstance("/", destMap); + entries.add(entry); + return entries; + } +} diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAdmin.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAdmin.java new file mode 100644 index 00000000000..170247ff8e6 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAdmin.java @@ -0,0 +1,261 @@ +/** + * 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.router; + +import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.synchronizeRecords; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import java.util.Collections; +import java.util.List; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hdfs.server.federation.RouterConfigBuilder; +import org.apache.hadoop.hdfs.server.federation.RouterDFSCluster.RouterContext; +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.RemoteLocation; +import org.apache.hadoop.hdfs.server.federation.store.StateStoreService; +import org.apache.hadoop.hdfs.server.federation.store.impl.MountTableStoreImpl; +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; +import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.RemoveMountTableEntryRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateMountTableEntryRequest; +import org.apache.hadoop.hdfs.server.federation.store.records.MountTable; +import org.apache.hadoop.util.Time; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +/** + * The administrator interface of the {@link Router} implemented by + * {@link RouterAdminServer}. + */ +public class TestRouterAdmin { + + private static StateStoreDFSCluster cluster; + private static RouterContext routerContext; + public static final String RPC_BEAN = + "Hadoop:service=Router,name=FederationRPC"; + private static List mockMountTable; + private static StateStoreService stateStore; + + @BeforeClass + public static void globalSetUp() throws Exception { + cluster = new StateStoreDFSCluster(false, 1); + // Build and start a router with State Store + admin + RPC + Configuration conf = new RouterConfigBuilder() + .stateStore() + .admin() + .rpc() + .build(); + cluster.addRouterOverrides(conf); + cluster.startRouters(); + routerContext = cluster.getRandomRouter(); + mockMountTable = cluster.generateMockMountTable(); + Router router = routerContext.getRouter(); + stateStore = router.getStateStore(); + } + + @AfterClass + public static void tearDown() { + cluster.stopRouter(routerContext); + } + + @Before + public void testSetup() throws Exception { + assertTrue( + synchronizeRecords(stateStore, mockMountTable, MountTable.class)); + } + + @Test + public void testAddMountTable() throws IOException { + MountTable newEntry = MountTable.newInstance( + "/testpath", Collections.singletonMap("ns0", "/testdir"), + Time.now(), Time.now()); + + RouterClient client = routerContext.getAdminClient(); + MountTableManager mountTable = client.getMountTableManager(); + + // Existing mount table size + List records = getMountTableEntries(mountTable); + assertEquals(records.size(), mockMountTable.size()); + + // Add + AddMountTableEntryRequest addRequest = + AddMountTableEntryRequest.newInstance(newEntry); + AddMountTableEntryResponse addResponse = + mountTable.addMountTableEntry(addRequest); + assertTrue(addResponse.getStatus()); + + // New mount table size + List records2 = getMountTableEntries(mountTable); + assertEquals(records2.size(), mockMountTable.size() + 1); + } + + @Test + public void testAddDuplicateMountTable() throws IOException { + MountTable newEntry = MountTable.newInstance("/testpath", + Collections.singletonMap("ns0", "/testdir"), Time.now(), Time.now()); + + RouterClient client = routerContext.getAdminClient(); + MountTableManager mountTable = client.getMountTableManager(); + + // Existing mount table size + List entries1 = getMountTableEntries(mountTable); + assertEquals(entries1.size(), mockMountTable.size()); + + // Add + AddMountTableEntryRequest addRequest = + AddMountTableEntryRequest.newInstance(newEntry); + AddMountTableEntryResponse addResponse = + mountTable.addMountTableEntry(addRequest); + assertTrue(addResponse.getStatus()); + + // New mount table size + List entries2 = getMountTableEntries(mountTable); + assertEquals(entries2.size(), mockMountTable.size() + 1); + + // Add again, should fail + AddMountTableEntryResponse addResponse2 = + mountTable.addMountTableEntry(addRequest); + assertFalse(addResponse2.getStatus()); + } + + @Test + public void testRemoveMountTable() throws IOException { + + RouterClient client = routerContext.getAdminClient(); + MountTableManager mountTable = client.getMountTableManager(); + + // Existing mount table size + List entries1 = getMountTableEntries(mountTable); + assertEquals(entries1.size(), mockMountTable.size()); + + // Remove an entry + RemoveMountTableEntryRequest removeRequest = + RemoveMountTableEntryRequest.newInstance("/"); + mountTable.removeMountTableEntry(removeRequest); + + // New mount table size + List entries2 = getMountTableEntries(mountTable); + assertEquals(entries2.size(), mockMountTable.size() - 1); + } + + @Test + public void testEditMountTable() throws IOException { + + RouterClient client = routerContext.getAdminClient(); + MountTableManager mountTable = client.getMountTableManager(); + + // Verify starting condition + MountTable entry = getMountTableEntry("/"); + assertEquals( + Collections.singletonList(new RemoteLocation("ns0", "/")), + entry.getDestinations()); + + // Edit the entry for / + MountTable updatedEntry = MountTable.newInstance( + "/", Collections.singletonMap("ns1", "/"), Time.now(), Time.now()); + UpdateMountTableEntryRequest updateRequest = + UpdateMountTableEntryRequest.newInstance(updatedEntry); + mountTable.updateMountTableEntry(updateRequest); + + // Verify edited condition + entry = getMountTableEntry("/"); + assertEquals( + Collections.singletonList(new RemoteLocation("ns1", "/")), + entry.getDestinations()); + } + + @Test + public void testGetMountTable() throws IOException { + + RouterClient client = routerContext.getAdminClient(); + MountTableManager mountTable = client.getMountTableManager(); + + // Verify size of table + List entries = getMountTableEntries(mountTable); + assertEquals(mockMountTable.size(), entries.size()); + + // Verify all entries are present + int matches = 0; + for (MountTable e : entries) { + for (MountTable entry : mockMountTable) { + assertEquals(e.getDestinations().size(), 1); + assertNotNull(e.getDateCreated()); + assertNotNull(e.getDateModified()); + if (entry.getSourcePath().equals(e.getSourcePath())) { + matches++; + } + } + } + assertEquals(matches, mockMountTable.size()); + } + + @Test + public void testGetSingleMountTableEntry() throws IOException { + MountTable entry = getMountTableEntry("/ns0"); + assertNotNull(entry); + assertEquals(entry.getSourcePath(), "/ns0"); + } + + /** + * Gets an existing mount table record in the state store. + * + * @param mount The mount point of the record to remove. + * @return The matching record if found, null if it is not found. + * @throws IOException If the state store could not be accessed. + */ + private MountTable getMountTableEntry(final String mount) throws IOException { + // Refresh the cache + stateStore.loadCache(MountTableStoreImpl.class, true); + + GetMountTableEntriesRequest request = + GetMountTableEntriesRequest.newInstance(mount); + RouterClient client = routerContext.getAdminClient(); + MountTableManager mountTable = client.getMountTableManager(); + List results = getMountTableEntries(mountTable, request); + if (results.size() > 0) { + // First result is sorted to have the shortest mount string length + return results.get(0); + } + return null; + } + + private List getMountTableEntries(MountTableManager mountTable) + throws IOException { + GetMountTableEntriesRequest request = + GetMountTableEntriesRequest.newInstance("/"); + return getMountTableEntries(mountTable, request); + } + + private List getMountTableEntries(MountTableManager mountTable, + GetMountTableEntriesRequest request) throws IOException { + stateStore.loadCache(MountTableStoreImpl.class, true); + GetMountTableEntriesResponse response = + mountTable.getMountTableEntries(request); + return response.getEntries(); + } +} \ No newline at end of file