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 we...@apache.org on 2018/03/20 06:26:32 UTC
[15/38] hadoop git commit: HDFS-13215. RBF: Move Router to its own
module.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/Router.java
----------------------------------------------------------------------
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
deleted file mode 100644
index be9adc1..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/Router.java
+++ /dev/null
@@ -1,656 +0,0 @@
-/**
- * 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.router.FederationUtil.newActiveNamenodeResolver;
-import static org.apache.hadoop.hdfs.server.federation.router.FederationUtil.newFileSubclusterResolver;
-
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.net.UnknownHostException;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hdfs.DFSConfigKeys;
-import org.apache.hadoop.hdfs.DFSUtil;
-import org.apache.hadoop.hdfs.HAUtil;
-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.FileSubclusterResolver;
-import org.apache.hadoop.hdfs.server.federation.store.RouterStore;
-import org.apache.hadoop.hdfs.server.federation.store.StateStoreService;
-import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
-import org.apache.hadoop.metrics2.source.JvmMetrics;
-import org.apache.hadoop.service.CompositeService;
-import org.apache.hadoop.util.JvmPauseMonitor;
-import org.apache.hadoop.util.Time;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.annotations.VisibleForTesting;
-
-/**
- * Router that provides a unified view of multiple federated HDFS clusters. It
- * has two main roles: (1) federated interface and (2) NameNode heartbeat.
- * <p>
- * For the federated interface, the Router receives a client request, checks the
- * State Store for the correct subcluster, and forwards the request to the
- * active Namenode of that subcluster. The reply from the Namenode then flows in
- * the opposite direction. The Routers are stateless and can be behind a load
- * balancer. HDFS clients connect to the router using the same interfaces as are
- * used to communicate with a namenode, namely the ClientProtocol RPC interface
- * and the WebHdfs HTTP interface exposed by the router. {@link RouterRpcServer}
- * {@link RouterHttpServer}
- * <p>
- * For NameNode heartbeat, the Router periodically checks the state of a
- * NameNode (usually on the same server) and reports their high availability
- * (HA) state and load/space status to the State Store. Note that this is an
- * optional role as a Router can be independent of any subcluster.
- * {@link StateStoreService} {@link NamenodeHeartbeatService}
- */
-@InterfaceAudience.Private
-@InterfaceStability.Evolving
-public class Router extends CompositeService {
-
- private static final Logger LOG = LoggerFactory.getLogger(Router.class);
-
-
- /** Configuration for the Router. */
- private Configuration conf;
-
- /** Router address/identifier. */
- private String routerId;
-
- /** RPC interface to the client. */
- private RouterRpcServer rpcServer;
- private InetSocketAddress rpcAddress;
-
- /** RPC interface for the admin. */
- private RouterAdminServer adminServer;
- private InetSocketAddress adminAddress;
-
- /** HTTP interface and web application. */
- private RouterHttpServer httpServer;
-
- /** Interface with the State Store. */
- private StateStoreService stateStore;
-
- /** Interface to map global name space to HDFS subcluster name spaces. */
- private FileSubclusterResolver subclusterResolver;
-
- /** Interface to identify the active NN for a nameservice or blockpool ID. */
- private ActiveNamenodeResolver namenodeResolver;
- /** Updates the namenode status in the namenode resolver. */
- private Collection<NamenodeHeartbeatService> namenodeHeartbeatServices;
-
- /** Router metrics. */
- private RouterMetricsService metrics;
-
- /** JVM pauses (GC and others). */
- private JvmPauseMonitor pauseMonitor;
-
- /** Quota usage update service. */
- private RouterQuotaUpdateService quotaUpdateService;
- /** Quota cache manager. */
- private RouterQuotaManager quotaManager;
-
- /** Manages the current state of the router. */
- private RouterStore routerStateManager;
- /** Heartbeat our run status to the router state manager. */
- private RouterHeartbeatService routerHeartbeatService;
- /** Enter/exit safemode. */
- private RouterSafemodeService safemodeService;
-
- /** The start time of the namesystem. */
- private final long startTime = Time.now();
-
- /** State of the Router. */
- private RouterServiceState state = RouterServiceState.UNINITIALIZED;
-
-
- /////////////////////////////////////////////////////////
- // Constructor
- /////////////////////////////////////////////////////////
-
- public Router() {
- super(Router.class.getName());
- }
-
- /////////////////////////////////////////////////////////
- // Service management
- /////////////////////////////////////////////////////////
-
- @Override
- protected void serviceInit(Configuration configuration) throws Exception {
- this.conf = configuration;
- updateRouterState(RouterServiceState.INITIALIZING);
-
- 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);
- if (this.namenodeResolver == null) {
- throw new IOException("Cannot find namenode resolver.");
- }
-
- // Lookup interface to map between the global and subcluster name spaces
- this.subclusterResolver = newFileSubclusterResolver(this.conf, this);
- if (this.subclusterResolver == null) {
- throw new IOException("Cannot find subcluster resolver");
- }
-
- if (conf.getBoolean(
- DFSConfigKeys.DFS_ROUTER_RPC_ENABLE,
- DFSConfigKeys.DFS_ROUTER_RPC_ENABLE_DEFAULT)) {
- // Create RPC server
- this.rpcServer = createRpcServer();
- addService(this.rpcServer);
- 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_HTTP_ENABLE,
- DFSConfigKeys.DFS_ROUTER_HTTP_ENABLE_DEFAULT)) {
- // Create HTTP server
- this.httpServer = createHttpServer();
- addService(this.httpServer);
- }
-
- if (conf.getBoolean(
- DFSConfigKeys.DFS_ROUTER_HEARTBEAT_ENABLE,
- DFSConfigKeys.DFS_ROUTER_HEARTBEAT_ENABLE_DEFAULT)) {
-
- // Create status updater for each monitored Namenode
- this.namenodeHeartbeatServices = createNamenodeHeartbeatServices();
- for (NamenodeHeartbeatService hearbeatService :
- this.namenodeHeartbeatServices) {
- addService(hearbeatService);
- }
-
- if (this.namenodeHeartbeatServices.isEmpty()) {
- LOG.error("Heartbeat is enabled but there are no namenodes to monitor");
- }
-
- // Periodically update the router state
- this.routerHeartbeatService = new RouterHeartbeatService(this);
- addService(this.routerHeartbeatService);
- }
-
- // Router metrics system
- if (conf.getBoolean(
- DFSConfigKeys.DFS_ROUTER_METRICS_ENABLE,
- DFSConfigKeys.DFS_ROUTER_METRICS_ENABLE_DEFAULT)) {
-
- DefaultMetricsSystem.initialize("Router");
-
- this.metrics = new RouterMetricsService(this);
- addService(this.metrics);
-
- // JVM pause monitor
- this.pauseMonitor = new JvmPauseMonitor();
- this.pauseMonitor.init(conf);
- }
-
- // Initial quota relevant service
- if (conf.getBoolean(DFSConfigKeys.DFS_ROUTER_QUOTA_ENABLE,
- DFSConfigKeys.DFS_ROUTER_QUOTA_ENABLED_DEFAULT)) {
- this.quotaManager = new RouterQuotaManager();
- this.quotaUpdateService = new RouterQuotaUpdateService(this);
- addService(this.quotaUpdateService);
- }
-
- // Safemode service to refuse RPC calls when the router is out of sync
- if (conf.getBoolean(
- DFSConfigKeys.DFS_ROUTER_SAFEMODE_ENABLE,
- DFSConfigKeys.DFS_ROUTER_SAFEMODE_ENABLE_DEFAULT)) {
- // Create safemode monitoring service
- this.safemodeService = new RouterSafemodeService(this);
- addService(this.safemodeService);
- }
-
- super.serviceInit(conf);
- }
-
- @Override
- protected void serviceStart() throws Exception {
-
- if (this.safemodeService == null) {
- // Router is running now
- updateRouterState(RouterServiceState.RUNNING);
- }
-
- if (this.pauseMonitor != null) {
- this.pauseMonitor.start();
- JvmMetrics jvmMetrics = this.metrics.getJvmMetrics();
- if (jvmMetrics != null) {
- jvmMetrics.setPauseMonitor(pauseMonitor);
- }
- }
-
- super.serviceStart();
- }
-
- @Override
- protected void serviceStop() throws Exception {
-
- // Update state
- updateRouterState(RouterServiceState.SHUTDOWN);
-
- // JVM pause monitor
- if (this.pauseMonitor != null) {
- this.pauseMonitor.stop();
- }
-
- super.serviceStop();
- }
-
- /**
- * Shutdown the router.
- */
- public void shutDown() {
- new Thread() {
- @Override
- public void run() {
- Router.this.stop();
- }
- }.start();
- }
-
- /////////////////////////////////////////////////////////
- // RPC Server
- /////////////////////////////////////////////////////////
-
- /**
- * Create a new Router RPC server to proxy ClientProtocol requests.
- *
- * @return New Router RPC Server.
- * @throws IOException If the router RPC server was not started.
- */
- protected RouterRpcServer createRpcServer() throws IOException {
- return new RouterRpcServer(this.conf, this, this.getNamenodeResolver(),
- this.getSubclusterResolver());
- }
-
- /**
- * Get the Router RPC server.
- *
- * @return Router RPC server.
- */
- public RouterRpcServer getRpcServer() {
- return this.rpcServer;
- }
-
- /**
- * Set the current RPC socket for the router.
- *
- * @param address RPC address.
- */
- protected void setRpcServerAddress(InetSocketAddress address) {
- this.rpcAddress = address;
-
- // Use the RPC address as our unique router Id
- if (this.rpcAddress != null) {
- try {
- String hostname = InetAddress.getLocalHost().getHostName();
- setRouterId(hostname + ":" + this.rpcAddress.getPort());
- } catch (UnknownHostException ex) {
- LOG.error("Cannot set unique router ID, address not resolvable {}",
- this.rpcAddress);
- }
- }
- }
-
- /**
- * Get the current RPC socket address for the router.
- *
- * @return InetSocketAddress
- */
- public InetSocketAddress getRpcServerAddress() {
- 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 address 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;
- }
-
- /////////////////////////////////////////////////////////
- // HTTP server
- /////////////////////////////////////////////////////////
-
- /**
- * Create an HTTP server for this Router.
- *
- * @return HTTP server for this Router.
- */
- protected RouterHttpServer createHttpServer() {
- return new RouterHttpServer(this);
- }
-
- /**
- * Get the current HTTP socket address for the router.
- *
- * @return InetSocketAddress HTTP address.
- */
- public InetSocketAddress getHttpServerAddress() {
- if (httpServer != null) {
- return httpServer.getHttpAddress();
- }
- return null;
- }
-
- /////////////////////////////////////////////////////////
- // Namenode heartbeat monitors
- /////////////////////////////////////////////////////////
-
- /**
- * Create each of the services that will monitor a Namenode.
- *
- * @return List of heartbeat services.
- */
- protected Collection<NamenodeHeartbeatService>
- createNamenodeHeartbeatServices() {
-
- Map<String, NamenodeHeartbeatService> ret = new HashMap<>();
-
- if (conf.getBoolean(
- DFSConfigKeys.DFS_ROUTER_MONITOR_LOCAL_NAMENODE,
- DFSConfigKeys.DFS_ROUTER_MONITOR_LOCAL_NAMENODE_DEFAULT)) {
- // Create a local heartbet service
- NamenodeHeartbeatService localHeartbeatService =
- createLocalNamenodeHearbeatService();
- if (localHeartbeatService != null) {
- String nnDesc = localHeartbeatService.getNamenodeDesc();
- ret.put(nnDesc, localHeartbeatService);
- }
- }
-
- // Create heartbeat services for a list specified by the admin
- String namenodes = this.conf.get(
- DFSConfigKeys.DFS_ROUTER_MONITOR_NAMENODE);
- if (namenodes != null) {
- for (String namenode : namenodes.split(",")) {
- String[] namenodeSplit = namenode.split("\\.");
- String nsId = null;
- String nnId = null;
- if (namenodeSplit.length == 2) {
- nsId = namenodeSplit[0];
- nnId = namenodeSplit[1];
- } else if (namenodeSplit.length == 1) {
- nsId = namenode;
- } else {
- LOG.error("Wrong Namenode to monitor: {}", namenode);
- }
- if (nsId != null) {
- NamenodeHeartbeatService heartbeatService =
- createNamenodeHearbeatService(nsId, nnId);
- if (heartbeatService != null) {
- ret.put(heartbeatService.getNamenodeDesc(), heartbeatService);
- }
- }
- }
- }
-
- return ret.values();
- }
-
- /**
- * Create a new status updater for the local Namenode.
- *
- * @return Updater of the status for the local Namenode.
- */
- protected NamenodeHeartbeatService createLocalNamenodeHearbeatService() {
- // Detect NN running in this machine
- String nsId = DFSUtil.getNamenodeNameServiceId(conf);
- String nnId = null;
- if (HAUtil.isHAEnabled(conf, nsId)) {
- nnId = HAUtil.getNameNodeId(conf, nsId);
- if (nnId == null) {
- LOG.error("Cannot find namenode id for local {}", nsId);
- }
- }
-
- return createNamenodeHearbeatService(nsId, nnId);
- }
-
- /**
- * Create a heartbeat monitor for a particular Namenode.
- *
- * @param nsId Identifier of the nameservice to monitor.
- * @param nnId Identifier of the namenode (HA) to monitor.
- * @return Updater of the status for the specified Namenode.
- */
- protected NamenodeHeartbeatService createNamenodeHearbeatService(
- String nsId, String nnId) {
-
- LOG.info("Creating heartbeat service for Namenode {} in {}", nnId, nsId);
- NamenodeHeartbeatService ret = new NamenodeHeartbeatService(
- namenodeResolver, nsId, nnId);
- return ret;
- }
-
- /////////////////////////////////////////////////////////
- // Router State Management
- /////////////////////////////////////////////////////////
-
- /**
- * Update the router state and heartbeat to the state store.
- *
- * @param state The new router state.
- */
- public void updateRouterState(RouterServiceState newState) {
- this.state = newState;
- if (this.routerHeartbeatService != null) {
- this.routerHeartbeatService.updateStateAsync();
- }
- }
-
- /**
- * Get the status of the router.
- *
- * @return Status of the router.
- */
- public RouterServiceState getRouterState() {
- return this.state;
- }
-
- /////////////////////////////////////////////////////////
- // Submodule getters
- /////////////////////////////////////////////////////////
-
- /**
- * Get the State Store service.
- *
- * @return State Store service.
- */
- public StateStoreService getStateStore() {
- return this.stateStore;
- }
-
- /**
- * Get the metrics system for the Router.
- *
- * @return Router metrics.
- */
- public RouterMetrics getRouterMetrics() {
- if (this.metrics != null) {
- return this.metrics.getRouterMetrics();
- }
- return null;
- }
-
- /**
- * Get the federation metrics.
- *
- * @return Federation metrics.
- */
- public FederationMetrics getMetrics() {
- if (this.metrics != null) {
- return this.metrics.getFederationMetrics();
- }
- return null;
- }
-
- /**
- * Get the subcluster resolver for files.
- *
- * @return Subcluster resolver for files.
- */
- public FileSubclusterResolver getSubclusterResolver() {
- return this.subclusterResolver;
- }
-
- /**
- * Get the namenode resolver for a subcluster.
- *
- * @return The namenode resolver for a subcluster.
- */
- public ActiveNamenodeResolver getNamenodeResolver() {
- return this.namenodeResolver;
- }
-
- /**
- * Get the state store interface for the router heartbeats.
- *
- * @return FederationRouterStateStore state store API handle.
- */
- public RouterStore getRouterStateManager() {
- if (this.routerStateManager == null && this.stateStore != null) {
- this.routerStateManager = this.stateStore.getRegisteredRecordStore(
- RouterStore.class);
- }
- return this.routerStateManager;
- }
-
- /////////////////////////////////////////////////////////
- // Router info
- /////////////////////////////////////////////////////////
-
- /**
- * Get the start date of the Router.
- *
- * @return Start date of the router.
- */
- public long getStartTime() {
- return this.startTime;
- }
-
- /**
- * Unique ID for the router, typically the hostname:port string for the
- * router's RPC server. This ID may be null on router startup before the RPC
- * server has bound to a port.
- *
- * @return Router identifier.
- */
- public String getRouterId() {
- return this.routerId;
- }
-
- /**
- * Sets a unique ID for this router.
- *
- * @param id Identifier of the Router.
- */
- public void setRouterId(String id) {
- this.routerId = id;
- if (this.stateStore != null) {
- this.stateStore.setIdentifier(this.routerId);
- }
- if (this.namenodeResolver != null) {
- this.namenodeResolver.setRouterId(this.routerId);
- }
- }
-
- /**
- * If the quota system is enabled in Router.
- */
- public boolean isQuotaEnabled() {
- return this.quotaManager != null;
- }
-
- /**
- * Get route quota manager.
- * @return RouterQuotaManager Quota manager.
- */
- public RouterQuotaManager getQuotaManager() {
- return this.quotaManager;
- }
-
- /**
- * Get quota cache update service.
- */
- @VisibleForTesting
- RouterQuotaUpdateService getQuotaCacheUpdateService() {
- return this.quotaUpdateService;
- }
-
- /**
- * Get the list of namenode heartbeat service.
- */
- @VisibleForTesting
- Collection<NamenodeHeartbeatService> getNamenodeHearbeatServices() {
- return this.namenodeHeartbeatServices;
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterAdminServer.java
----------------------------------------------------------------------
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
deleted file mode 100644
index 0c0448c..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterAdminServer.java
+++ /dev/null
@@ -1,298 +0,0 @@
-/**
- * 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.DFSConfigKeys.DFS_PERMISSIONS_ENABLED_DEFAULT;
-import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_PERMISSIONS_ENABLED_KEY;
-
-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.EnterSafeModeRequest;
-import org.apache.hadoop.hdfs.server.federation.store.protocol.EnterSafeModeResponse;
-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;
-import org.apache.hadoop.hdfs.server.federation.store.protocol.GetSafeModeResponse;
-import org.apache.hadoop.hdfs.server.federation.store.protocol.LeaveSafeModeRequest;
-import org.apache.hadoop.hdfs.server.federation.store.protocol.LeaveSafeModeResponse;
-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.namenode.NameNode;
-import org.apache.hadoop.ipc.ProtobufRpcEngine;
-import org.apache.hadoop.ipc.RPC;
-import org.apache.hadoop.ipc.RPC.Server;
-import org.apache.hadoop.security.AccessControlException;
-import org.apache.hadoop.security.UserGroupInformation;
-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, RouterStateManager {
-
- 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;
-
- /**
- * Permission related info used for constructing new router permission
- * checker instance.
- */
- private static String routerOwner;
- private static String superGroup;
- private static boolean isPermissionEnabled;
-
- 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());
-
- initializePermissionSettings(this.conf);
- 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);
- }
-
- /**
- * Initialize permission related settings.
- *
- * @param routerConf
- * @throws IOException
- */
- private static void initializePermissionSettings(Configuration routerConf)
- throws IOException {
- routerOwner = UserGroupInformation.getCurrentUser().getShortUserName();
- superGroup = routerConf.get(
- DFSConfigKeys.DFS_PERMISSIONS_SUPERUSERGROUP_KEY,
- DFSConfigKeys.DFS_PERMISSIONS_SUPERUSERGROUP_DEFAULT);
- isPermissionEnabled = routerConf.getBoolean(DFS_PERMISSIONS_ENABLED_KEY,
- DFS_PERMISSIONS_ENABLED_DEFAULT);
- }
-
- /** 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);
- }
-
- @Override
- public EnterSafeModeResponse enterSafeMode(EnterSafeModeRequest request)
- throws IOException {
- this.router.updateRouterState(RouterServiceState.SAFEMODE);
- this.router.getRpcServer().setSafeMode(true);
- return EnterSafeModeResponse.newInstance(verifySafeMode(true));
- }
-
- @Override
- public LeaveSafeModeResponse leaveSafeMode(LeaveSafeModeRequest request)
- throws IOException {
- this.router.updateRouterState(RouterServiceState.RUNNING);
- this.router.getRpcServer().setSafeMode(false);
- return LeaveSafeModeResponse.newInstance(verifySafeMode(false));
- }
-
- @Override
- public GetSafeModeResponse getSafeMode(GetSafeModeRequest request)
- throws IOException {
- boolean isInSafeMode = this.router.getRpcServer().isInSafeMode();
- return GetSafeModeResponse.newInstance(isInSafeMode);
- }
-
- /**
- * Verify if Router set safe mode state correctly.
- * @param isInSafeMode Expected state to be set.
- * @return
- */
- private boolean verifySafeMode(boolean isInSafeMode) {
- boolean serverInSafeMode = this.router.getRpcServer().isInSafeMode();
- RouterServiceState currentState = this.router.getRouterState();
-
- return (isInSafeMode && currentState == RouterServiceState.SAFEMODE
- && serverInSafeMode)
- || (!isInSafeMode && currentState != RouterServiceState.SAFEMODE
- && !serverInSafeMode);
- }
-
- /**
- * Get a new permission checker used for making mount table access
- * control. This method will be invoked during each RPC call in router
- * admin server.
- *
- * @return Router permission checker
- * @throws AccessControlException
- */
- public static RouterPermissionChecker getPermissionChecker()
- throws AccessControlException {
- if (!isPermissionEnabled) {
- return null;
- }
-
- try {
- return new RouterPermissionChecker(routerOwner, superGroup,
- NameNode.getRemoteUser());
- } catch (IOException e) {
- throw new AccessControlException(e);
- }
- }
-
- /**
- * Get super user name.
- *
- * @return String super user name.
- */
- public static String getSuperUser() {
- return routerOwner;
- }
-
- /**
- * Get super group name.
- *
- * @return String super group name.
- */
- public static String getSuperGroup(){
- return superGroup;
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterClient.java
----------------------------------------------------------------------
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
deleted file mode 100644
index b36e459..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterClient.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/**
- * 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;
- }
-
- public RouterStateManager getRouterStateManager() {
- return proxy;
- }
-
- @Override
- public synchronized void close() throws IOException {
- RPC.stopProxy(proxy);
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterHeartbeatService.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterHeartbeatService.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterHeartbeatService.java
deleted file mode 100644
index 6e44984..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterHeartbeatService.java
+++ /dev/null
@@ -1,167 +0,0 @@
-/**
- * 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.util.List;
-import java.util.concurrent.TimeUnit;
-
-import com.google.common.annotations.VisibleForTesting;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hdfs.DFSConfigKeys;
-import org.apache.hadoop.hdfs.server.federation.store.CachedRecordStore;
-import org.apache.hadoop.hdfs.server.federation.store.MembershipStore;
-import org.apache.hadoop.hdfs.server.federation.store.MountTableStore;
-import org.apache.hadoop.hdfs.server.federation.store.RecordStore;
-import org.apache.hadoop.hdfs.server.federation.store.RouterStore;
-import org.apache.hadoop.hdfs.server.federation.store.StateStoreService;
-import org.apache.hadoop.hdfs.server.federation.store.protocol.RouterHeartbeatRequest;
-import org.apache.hadoop.hdfs.server.federation.store.protocol.RouterHeartbeatResponse;
-import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord;
-import org.apache.hadoop.hdfs.server.federation.store.records.RouterState;
-import org.apache.hadoop.hdfs.server.federation.store.records.StateStoreVersion;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Service to periodically update the Router current state in the State Store.
- */
-public class RouterHeartbeatService extends PeriodicService {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(RouterHeartbeatService.class);
-
- /** Router we are hearbeating. */
- private final Router router;
-
- /**
- * Create a new Router heartbeat service.
- *
- * @param router Router to heartbeat.
- */
- public RouterHeartbeatService(Router router) {
- super(RouterHeartbeatService.class.getSimpleName());
- this.router = router;
- }
-
- /**
- * Trigger the update of the Router state asynchronously.
- */
- protected void updateStateAsync() {
- Thread thread = new Thread(new Runnable() {
- @Override
- public void run() {
- updateStateStore();
- }
- }, "Router Heartbeat Async");
- thread.setDaemon(true);
- thread.start();
- }
-
- /**
- * Update the state of the Router in the State Store.
- */
- @VisibleForTesting
- synchronized void updateStateStore() {
- String routerId = router.getRouterId();
- if (routerId == null) {
- LOG.error("Cannot heartbeat for router: unknown router id");
- return;
- }
- if (isStoreAvailable()) {
- RouterStore routerStore = router.getRouterStateManager();
- try {
- RouterState record = RouterState.newInstance(
- routerId, router.getStartTime(), router.getRouterState());
- StateStoreVersion stateStoreVersion = StateStoreVersion.newInstance(
- getStateStoreVersion(MembershipStore.class),
- getStateStoreVersion(MountTableStore.class));
- record.setStateStoreVersion(stateStoreVersion);
- RouterHeartbeatRequest request =
- RouterHeartbeatRequest.newInstance(record);
- RouterHeartbeatResponse response = routerStore.routerHeartbeat(request);
- if (!response.getStatus()) {
- LOG.warn("Cannot heartbeat router {}", routerId);
- } else {
- LOG.debug("Router heartbeat for router {}", routerId);
- }
- } catch (IOException e) {
- LOG.error("Cannot heartbeat router {}: {}", routerId, e.getMessage());
- }
- } else {
- LOG.warn("Cannot heartbeat router {}: State Store unavailable", routerId);
- }
- }
-
- /**
- * Get the version of the data in the State Store.
- *
- * @param clazz Class in the State Store.
- * @return Version of the data.
- */
- private <R extends BaseRecord, S extends RecordStore<R>>
- long getStateStoreVersion(final Class<S> clazz) {
- long version = -1;
- try {
- StateStoreService stateStore = router.getStateStore();
- S recordStore = stateStore.getRegisteredRecordStore(clazz);
- if (recordStore != null) {
- if (recordStore instanceof CachedRecordStore) {
- CachedRecordStore<R> cachedRecordStore =
- (CachedRecordStore<R>) recordStore;
- List<R> records = cachedRecordStore.getCachedRecords();
- for (BaseRecord record : records) {
- if (record.getDateModified() > version) {
- version = record.getDateModified();
- }
- }
- }
- }
- } catch (Exception e) {
- LOG.error("Cannot get version for {}: {}", clazz, e.getMessage());
- }
- return version;
- }
-
- @Override
- protected void serviceInit(Configuration conf) throws Exception {
-
- long interval = conf.getTimeDuration(
- DFSConfigKeys.DFS_ROUTER_HEARTBEAT_STATE_INTERVAL_MS,
- DFSConfigKeys.DFS_ROUTER_HEARTBEAT_STATE_INTERVAL_MS_DEFAULT,
- TimeUnit.MILLISECONDS);
- this.setIntervalMs(interval);
-
- super.serviceInit(conf);
- }
-
- @Override
- public void periodicInvoke() {
- updateStateStore();
- }
-
- private boolean isStoreAvailable() {
- if (router.getRouterStateManager() == null) {
- return false;
- }
- if (router.getStateStore() == null) {
- return false;
- }
- return router.getStateStore().isDriverReady();
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterHttpServer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterHttpServer.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterHttpServer.java
deleted file mode 100644
index 046f0ba..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterHttpServer.java
+++ /dev/null
@@ -1,124 +0,0 @@
-/**
- * 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.net.InetSocketAddress;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hdfs.DFSConfigKeys;
-import org.apache.hadoop.hdfs.DFSUtil;
-import org.apache.hadoop.hdfs.server.common.JspHelper;
-import org.apache.hadoop.http.HttpServer2;
-import org.apache.hadoop.service.AbstractService;
-
-/**
- * Web interface for the {@link Router}. It exposes the Web UI and the WebHDFS
- * methods from {@link RouterWebHdfsMethods}.
- */
-public class RouterHttpServer extends AbstractService {
-
- protected static final String NAMENODE_ATTRIBUTE_KEY = "name.node";
-
-
- /** Configuration for the Router HTTP server. */
- private Configuration conf;
-
- /** Router using this HTTP server. */
- private final Router router;
-
- /** HTTP server. */
- private HttpServer2 httpServer;
-
- /** HTTP addresses. */
- private InetSocketAddress httpAddress;
- private InetSocketAddress httpsAddress;
-
-
- public RouterHttpServer(Router router) {
- super(RouterHttpServer.class.getName());
- this.router = router;
- }
-
- @Override
- protected void serviceInit(Configuration configuration) throws Exception {
- this.conf = configuration;
-
- // Get HTTP address
- this.httpAddress = conf.getSocketAddr(
- DFSConfigKeys.DFS_ROUTER_HTTP_BIND_HOST_KEY,
- DFSConfigKeys.DFS_ROUTER_HTTP_ADDRESS_KEY,
- DFSConfigKeys.DFS_ROUTER_HTTP_ADDRESS_DEFAULT,
- DFSConfigKeys.DFS_ROUTER_HTTP_PORT_DEFAULT);
-
- // Get HTTPs address
- this.httpsAddress = conf.getSocketAddr(
- DFSConfigKeys.DFS_ROUTER_HTTPS_BIND_HOST_KEY,
- DFSConfigKeys.DFS_ROUTER_HTTPS_ADDRESS_KEY,
- DFSConfigKeys.DFS_ROUTER_HTTPS_ADDRESS_DEFAULT,
- DFSConfigKeys.DFS_ROUTER_HTTPS_PORT_DEFAULT);
-
- super.serviceInit(conf);
- }
-
- @Override
- protected void serviceStart() throws Exception {
- // Build and start server
- String webApp = "router";
- HttpServer2.Builder builder = DFSUtil.httpServerTemplateForNNAndJN(
- this.conf, this.httpAddress, this.httpsAddress, webApp,
- DFSConfigKeys.DFS_NAMENODE_KERBEROS_INTERNAL_SPNEGO_PRINCIPAL_KEY,
- DFSConfigKeys.DFS_NAMENODE_KEYTAB_FILE_KEY);
-
- this.httpServer = builder.build();
-
- this.httpServer.setAttribute(NAMENODE_ATTRIBUTE_KEY, this.router);
- this.httpServer.setAttribute(JspHelper.CURRENT_CONF, this.conf);
- setupServlets(this.httpServer, this.conf);
-
- this.httpServer.start();
-
- // The server port can be ephemeral... ensure we have the correct info
- InetSocketAddress listenAddress = this.httpServer.getConnectorAddress(0);
- if (listenAddress != null) {
- this.httpAddress = new InetSocketAddress(this.httpAddress.getHostName(),
- listenAddress.getPort());
- }
- super.serviceStart();
- }
-
- @Override
- protected void serviceStop() throws Exception {
- if(this.httpServer != null) {
- this.httpServer.stop();
- }
- super.serviceStop();
- }
-
- private static void setupServlets(
- HttpServer2 httpServer, Configuration conf) {
- // TODO Add servlets for FSCK, etc
- }
-
- public InetSocketAddress getHttpAddress() {
- return this.httpAddress;
- }
-
- public InetSocketAddress getHttpsAddress() {
- return this.httpsAddress;
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterMetrics.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterMetrics.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterMetrics.java
deleted file mode 100644
index 851538a..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterMetrics.java
+++ /dev/null
@@ -1,73 +0,0 @@
-/**
- * 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.metrics2.impl.MsInfo.ProcessName;
-import static org.apache.hadoop.metrics2.impl.MsInfo.SessionId;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hdfs.DFSConfigKeys;
-import org.apache.hadoop.metrics2.MetricsSystem;
-import org.apache.hadoop.metrics2.annotation.Metric;
-import org.apache.hadoop.metrics2.annotation.Metrics;
-import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
-import org.apache.hadoop.metrics2.lib.MetricsRegistry;
-import org.apache.hadoop.metrics2.lib.MutableGaugeInt;
-import org.apache.hadoop.metrics2.source.JvmMetrics;
-
-/**
- * This class is for maintaining the various Router activity statistics
- * and publishing them through the metrics interfaces.
- */
-@Metrics(name="RouterActivity", about="Router metrics", context="dfs")
-public class RouterMetrics {
-
- private final MetricsRegistry registry = new MetricsRegistry("router");
-
- @Metric("Duration in SafeMode at startup in msec")
- private MutableGaugeInt safeModeTime;
-
- private JvmMetrics jvmMetrics = null;
-
- RouterMetrics(
- String processName, String sessionId, final JvmMetrics jvmMetrics) {
- this.jvmMetrics = jvmMetrics;
- registry.tag(ProcessName, processName).tag(SessionId, sessionId);
- }
-
- public static RouterMetrics create(Configuration conf) {
- String sessionId = conf.get(DFSConfigKeys.DFS_METRICS_SESSION_ID_KEY);
- String processName = "Router";
- MetricsSystem ms = DefaultMetricsSystem.instance();
- JvmMetrics jm = JvmMetrics.create(processName, sessionId, ms);
-
- return ms.register(new RouterMetrics(processName, sessionId, jm));
- }
-
- public JvmMetrics getJvmMetrics() {
- return jvmMetrics;
- }
-
- public void shutdown() {
- DefaultMetricsSystem.shutdown();
- }
-
- public void setSafeModeTime(long elapsed) {
- safeModeTime.set((int) elapsed);
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterMetricsService.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterMetricsService.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterMetricsService.java
deleted file mode 100644
index f4debce..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterMetricsService.java
+++ /dev/null
@@ -1,108 +0,0 @@
-/**
- * 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 org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hdfs.server.federation.metrics.FederationMetrics;
-import org.apache.hadoop.hdfs.server.federation.metrics.NamenodeBeanMetrics;
-import org.apache.hadoop.metrics2.source.JvmMetrics;
-import org.apache.hadoop.service.AbstractService;
-
-/**
- * Service to manage the metrics of the Router.
- */
-public class RouterMetricsService extends AbstractService {
-
- /** Router for this metrics. */
- private final Router router;
-
- /** Router metrics. */
- private RouterMetrics routerMetrics;
- /** Federation metrics. */
- private FederationMetrics federationMetrics;
- /** Namenode mock metrics. */
- private NamenodeBeanMetrics nnMetrics;
-
-
- public RouterMetricsService(final Router router) {
- super(RouterMetricsService.class.getName());
- this.router = router;
- }
-
- @Override
- protected void serviceInit(Configuration configuration) throws Exception {
- this.routerMetrics = RouterMetrics.create(configuration);
- }
-
- @Override
- protected void serviceStart() throws Exception {
- // Wrapper for all the FSNamesystem JMX interfaces
- this.nnMetrics = new NamenodeBeanMetrics(this.router);
-
- // Federation MBean JMX interface
- this.federationMetrics = new FederationMetrics(this.router);
- }
-
- @Override
- protected void serviceStop() throws Exception {
- // Remove JMX interfaces
- if (this.federationMetrics != null) {
- this.federationMetrics.close();
- }
-
- // Remove Namenode JMX interfaces
- if (this.nnMetrics != null) {
- this.nnMetrics.close();
- }
-
- // Shutdown metrics
- if (this.routerMetrics != null) {
- this.routerMetrics.shutdown();
- }
- }
-
- /**
- * Get the metrics system for the Router.
- *
- * @return Router metrics.
- */
- public RouterMetrics getRouterMetrics() {
- return this.routerMetrics;
- }
-
- /**
- * Get the federation metrics.
- *
- * @return Federation metrics.
- */
- public FederationMetrics getFederationMetrics() {
- return this.federationMetrics;
- }
-
- /**
- * Get the JVM metrics for the Router.
- *
- * @return JVM metrics.
- */
- public JvmMetrics getJvmMetrics() {
- if (this.routerMetrics == null) {
- return null;
- }
- return this.routerMetrics.getJvmMetrics();
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterPermissionChecker.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterPermissionChecker.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterPermissionChecker.java
deleted file mode 100644
index 9d81dce..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterPermissionChecker.java
+++ /dev/null
@@ -1,82 +0,0 @@
-/**
- * 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 org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.fs.permission.FsAction;
-import org.apache.hadoop.fs.permission.FsPermission;
-import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
-import org.apache.hadoop.hdfs.server.namenode.FSPermissionChecker;
-import org.apache.hadoop.security.AccessControlException;
-import org.apache.hadoop.security.UserGroupInformation;
-
-/**
- * Class that helps in checking permissions in Router-based federation.
- */
-public class RouterPermissionChecker extends FSPermissionChecker {
- static final Log LOG = LogFactory.getLog(RouterPermissionChecker.class);
-
- /** Mount table default permission. */
- public static final short MOUNT_TABLE_PERMISSION_DEFAULT = 00755;
-
- public RouterPermissionChecker(String routerOwner, String supergroup,
- UserGroupInformation callerUgi) {
- super(routerOwner, supergroup, callerUgi, null);
- }
-
- /**
- * Whether a mount table entry can be accessed by the current context.
- *
- * @param mountTable
- * MountTable being accessed
- * @param access
- * type of action being performed on the cache pool
- * @throws AccessControlException
- * if mount table cannot be accessed
- */
- public void checkPermission(MountTable mountTable, FsAction access)
- throws AccessControlException {
- if (isSuperUser()) {
- return;
- }
-
- FsPermission mode = mountTable.getMode();
- if (getUser().equals(mountTable.getOwnerName())
- && mode.getUserAction().implies(access)) {
- return;
- }
-
- if (isMemberOfGroup(mountTable.getGroupName())
- && mode.getGroupAction().implies(access)) {
- return;
- }
-
- if (!getUser().equals(mountTable.getOwnerName())
- && !isMemberOfGroup(mountTable.getGroupName())
- && mode.getOtherAction().implies(access)) {
- return;
- }
-
- throw new AccessControlException(
- "Permission denied while accessing mount table "
- + mountTable.getSourcePath()
- + ": user " + getUser() + " does not have " + access.toString()
- + " permissions.");
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterQuotaManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterQuotaManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterQuotaManager.java
deleted file mode 100644
index 0df34fc..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterQuotaManager.java
+++ /dev/null
@@ -1,172 +0,0 @@
-/**
- * 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.router.FederationUtil.isParentEntry;
-
-import java.util.HashSet;
-import java.util.Set;
-import java.util.SortedMap;
-import java.util.TreeMap;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hdfs.protocol.HdfsConstants;
-
-/**
- * Router quota manager in Router. The manager maintains
- * {@link RouterQuotaUsage} cache of mount tables and do management
- * for the quota caches.
- */
-public class RouterQuotaManager {
- /** Quota usage <MountTable Path, Aggregated QuotaUsage> cache. */
- private TreeMap<String, RouterQuotaUsage> cache;
-
- /** Lock to access the quota cache. */
- private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
- private final Lock readLock = readWriteLock.readLock();
- private final Lock writeLock = readWriteLock.writeLock();
-
- public RouterQuotaManager() {
- this.cache = new TreeMap<>();
- }
-
- /**
- * Get all the mount quota paths.
- */
- public Set<String> getAll() {
- readLock.lock();
- try {
- return this.cache.keySet();
- } finally {
- readLock.unlock();
- }
- }
-
- /**
- * Get the nearest ancestor's quota usage, and meanwhile its quota was set.
- * @param path The path being written.
- * @return RouterQuotaUsage Quota usage.
- */
- public RouterQuotaUsage getQuotaUsage(String path) {
- readLock.lock();
- try {
- RouterQuotaUsage quotaUsage = this.cache.get(path);
- if (quotaUsage != null && isQuotaSet(quotaUsage)) {
- return quotaUsage;
- }
-
- // If not found, look for its parent path usage value.
- int pos = path.lastIndexOf(Path.SEPARATOR);
- if (pos != -1) {
- String parentPath = path.substring(0, pos);
- return getQuotaUsage(parentPath);
- }
- } finally {
- readLock.unlock();
- }
-
- return null;
- }
-
- /**
- * Get children paths (can including itself) under specified federation path.
- * @param parentPath
- * @return Set<String> Children path set.
- */
- public Set<String> getPaths(String parentPath) {
- readLock.lock();
- try {
- String from = parentPath;
- String to = parentPath + Character.MAX_VALUE;
- SortedMap<String, RouterQuotaUsage> subMap = this.cache.subMap(from, to);
-
- Set<String> validPaths = new HashSet<>();
- if (subMap != null) {
- for (String path : subMap.keySet()) {
- if (isParentEntry(path, parentPath)) {
- validPaths.add(path);
- }
- }
- }
- return validPaths;
- } finally {
- readLock.unlock();
- }
- }
-
- /**
- * Put new entity into cache.
- * @param path Mount table path.
- * @param quotaUsage Corresponding cache value.
- */
- public void put(String path, RouterQuotaUsage quotaUsage) {
- writeLock.lock();
- try {
- this.cache.put(path, quotaUsage);
- } finally {
- writeLock.unlock();
- }
- }
-
- /**
- * Remove the entity from cache.
- * @param path Mount table path.
- */
- public void remove(String path) {
- writeLock.lock();
- try {
- this.cache.remove(path);
- } finally {
- writeLock.unlock();
- }
- }
-
- /**
- * Clean up the cache.
- */
- public void clear() {
- writeLock.lock();
- try {
- this.cache.clear();
- } finally {
- writeLock.unlock();
- }
- }
-
- /**
- * Check if the quota was set.
- * @param quota RouterQuotaUsage set in mount table.
- */
- public boolean isQuotaSet(RouterQuotaUsage quota) {
- if (quota != null) {
- long nsQuota = quota.getQuota();
- long ssQuota = quota.getSpaceQuota();
-
- // once nsQuota or ssQuota was set, this mount table is quota set
- if (nsQuota != HdfsConstants.QUOTA_DONT_SET
- || ssQuota != HdfsConstants.QUOTA_DONT_SET) {
- return true;
- }
- }
-
- return false;
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterQuotaUpdateService.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterQuotaUpdateService.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterQuotaUpdateService.java
deleted file mode 100644
index 80abc11..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterQuotaUpdateService.java
+++ /dev/null
@@ -1,228 +0,0 @@
-/**
- * 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.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.QuotaUsage;
-import org.apache.hadoop.hdfs.DFSConfigKeys;
-import org.apache.hadoop.hdfs.protocol.HdfsConstants;
-import org.apache.hadoop.hdfs.server.federation.store.MountTableStore;
-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.UpdateMountTableEntryRequest;
-import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Service to periodically update the {@link RouterQuotaUsage}
- * cached information in the {@link Router} and update corresponding
- * mount table in State Store.
- */
-public class RouterQuotaUpdateService extends PeriodicService {
- private static final Logger LOG =
- LoggerFactory.getLogger(RouterQuotaUpdateService.class);
-
- private MountTableStore mountTableStore;
- private RouterRpcServer rpcServer;
- /** Router using this Service. */
- private final Router router;
- /** Router Quota manager. */
- private RouterQuotaManager quotaManager;
-
- public RouterQuotaUpdateService(final Router router) throws IOException {
- super(RouterQuotaUpdateService.class.getName());
- this.router = router;
- this.rpcServer = router.getRpcServer();
- this.quotaManager = router.getQuotaManager();
-
- if (this.quotaManager == null) {
- throw new IOException("Router quota manager is not initialized.");
- }
- }
-
- @Override
- protected void serviceInit(Configuration conf) throws Exception {
- this.setIntervalMs(conf.getTimeDuration(
- DFSConfigKeys.DFS_ROUTER_QUOTA_CACHE_UPATE_INTERVAL,
- DFSConfigKeys.DFS_ROUTER_QUOTA_CACHE_UPATE_INTERVAL_DEFAULT,
- TimeUnit.MILLISECONDS));
-
- super.serviceInit(conf);
- }
-
- @Override
- protected void periodicInvoke() {
- LOG.debug("Start to update quota cache.");
- try {
- List<MountTable> updateMountTables = new LinkedList<>();
- List<MountTable> mountTables = getQuotaSetMountTables();
- for (MountTable entry : mountTables) {
- String src = entry.getSourcePath();
- RouterQuotaUsage oldQuota = entry.getQuota();
- long nsQuota = oldQuota.getQuota();
- long ssQuota = oldQuota.getSpaceQuota();
- // Call RouterRpcServer#getQuotaUsage for getting current quota usage.
- QuotaUsage currentQuotaUsage = this.rpcServer.getQuotaModule()
- .getQuotaUsage(src);
- // If quota is not set in some subclusters under federation path,
- // set quota for this path.
- if (currentQuotaUsage.getQuota() == HdfsConstants.QUOTA_DONT_SET) {
- this.rpcServer.setQuota(src, nsQuota, ssQuota, null);
- }
-
- RouterQuotaUsage newQuota = generateNewQuota(oldQuota,
- currentQuotaUsage);
- this.quotaManager.put(src, newQuota);
- entry.setQuota(newQuota);
-
- // only update mount tables which quota was changed
- if (!oldQuota.equals(newQuota)) {
- updateMountTables.add(entry);
-
- LOG.debug(
- "Update quota usage entity of path: {}, nsCount: {},"
- + " nsQuota: {}, ssCount: {}, ssQuota: {}.",
- src, newQuota.getFileAndDirectoryCount(),
- newQuota.getQuota(), newQuota.getSpaceConsumed(),
- newQuota.getSpaceQuota());
- }
- }
-
- updateMountTableEntries(updateMountTables);
- } catch (IOException e) {
- LOG.error("Quota cache updated error.", e);
- }
- }
-
- /**
- * Get mount table store management interface.
- * @return MountTableStore instance.
- * @throws IOException
- */
- 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 all the existing mount tables.
- * @return List of mount tables.
- * @throws IOException
- */
- private List<MountTable> getMountTableEntries() throws IOException {
- // scan mount tables from root path
- GetMountTableEntriesRequest getRequest = GetMountTableEntriesRequest
- .newInstance("/");
- GetMountTableEntriesResponse getResponse = getMountTableStore()
- .getMountTableEntries(getRequest);
- return getResponse.getEntries();
- }
-
- /**
- * Get mount tables which quota was set.
- * During this time, the quota usage cache will also be updated by
- * quota manager:
- * 1. Stale paths (entries) will be removed.
- * 2. Existing entries will be override and updated.
- * @return List of mount tables which quota was set.
- * @throws IOException
- */
- private List<MountTable> getQuotaSetMountTables() throws IOException {
- List<MountTable> mountTables = getMountTableEntries();
- Set<String> stalePaths = new HashSet<>();
- for (String path : this.quotaManager.getAll()) {
- stalePaths.add(path);
- }
-
- List<MountTable> neededMountTables = new LinkedList<>();
- for (MountTable entry : mountTables) {
- // select mount tables which is quota set
- if (isQuotaSet(entry)) {
- neededMountTables.add(entry);
- }
-
- // update mount table entries info in quota cache
- String src = entry.getSourcePath();
- this.quotaManager.put(src, entry.getQuota());
- stalePaths.remove(src);
- }
-
- // remove stale paths that currently cached
- for (String stalePath : stalePaths) {
- this.quotaManager.remove(stalePath);
- }
-
- return neededMountTables;
- }
-
- /**
- * Check if the quota was set in given MountTable.
- * @param mountTable Mount table entry.
- */
- private boolean isQuotaSet(MountTable mountTable) {
- if (mountTable != null) {
- return this.quotaManager.isQuotaSet(mountTable.getQuota());
- }
- return false;
- }
-
- /**
- * Generate a new quota based on old quota and current quota usage value.
- * @param oldQuota Old quota stored in State Store.
- * @param currentQuotaUsage Current quota usage value queried from
- * subcluster.
- * @return A new RouterQuotaUsage.
- */
- private RouterQuotaUsage generateNewQuota(RouterQuotaUsage oldQuota,
- QuotaUsage currentQuotaUsage) {
- RouterQuotaUsage newQuota = new RouterQuotaUsage.Builder()
- .fileAndDirectoryCount(currentQuotaUsage.getFileAndDirectoryCount())
- .quota(oldQuota.getQuota())
- .spaceConsumed(currentQuotaUsage.getSpaceConsumed())
- .spaceQuota(oldQuota.getSpaceQuota()).build();
- return newQuota;
- }
-
- /**
- * Write out updated mount table entries into State Store.
- * @param updateMountTables Mount tables to be updated.
- * @throws IOException
- */
- private void updateMountTableEntries(List<MountTable> updateMountTables)
- throws IOException {
- for (MountTable entry : updateMountTables) {
- UpdateMountTableEntryRequest updateRequest = UpdateMountTableEntryRequest
- .newInstance(entry);
- getMountTableStore().updateMountTableEntry(updateRequest);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterQuotaUsage.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterQuotaUsage.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterQuotaUsage.java
deleted file mode 100644
index eedd80f..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterQuotaUsage.java
+++ /dev/null
@@ -1,119 +0,0 @@
-/**
- * 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 org.apache.hadoop.fs.QuotaUsage;
-import org.apache.hadoop.hdfs.protocol.DSQuotaExceededException;
-import org.apache.hadoop.hdfs.protocol.HdfsConstants;
-import org.apache.hadoop.hdfs.protocol.NSQuotaExceededException;
-import org.apache.hadoop.hdfs.server.namenode.DirectoryWithQuotaFeature;
-import org.apache.hadoop.hdfs.server.namenode.Quota;
-import org.apache.hadoop.util.StringUtils;
-
-/**
- * The subclass of {@link QuotaUsage} used in Router-based federation.
- */
-public final class RouterQuotaUsage extends QuotaUsage {
-
- /** Default quota usage count. */
- public static final long QUOTA_USAGE_COUNT_DEFAULT = 0;
-
- private RouterQuotaUsage(Builder builder) {
- super(builder);
- }
-
- /** Build the instance based on the builder. */
- public static class Builder extends QuotaUsage.Builder {
-
- public RouterQuotaUsage build() {
- return new RouterQuotaUsage(this);
- }
-
- @Override
- public Builder fileAndDirectoryCount(long count) {
- super.fileAndDirectoryCount(count);
- return this;
- }
-
- @Override
- public Builder quota(long quota) {
- super.quota(quota);
- return this;
- }
-
- @Override
- public Builder spaceConsumed(long spaceConsumed) {
- super.spaceConsumed(spaceConsumed);
- return this;
- }
-
- @Override
- public Builder spaceQuota(long spaceQuota) {
- super.spaceQuota(spaceQuota);
- return this;
- }
- }
-
- /**
- * Verify if namespace quota is violated once quota is set. Relevant
- * method {@link DirectoryWithQuotaFeature#verifyNamespaceQuota}.
- * @throws NSQuotaExceededException
- */
- public void verifyNamespaceQuota() throws NSQuotaExceededException {
- if (Quota.isViolated(getQuota(), getFileAndDirectoryCount())) {
- throw new NSQuotaExceededException(getQuota(),
- getFileAndDirectoryCount());
- }
- }
-
- /**
- * Verify if storage space quota is violated once quota is set. Relevant
- * method {@link DirectoryWithQuotaFeature#verifyStoragespaceQuota}.
- * @throws DSQuotaExceededException
- */
- public void verifyStoragespaceQuota() throws DSQuotaExceededException {
- if (Quota.isViolated(getSpaceQuota(), getSpaceConsumed())) {
- throw new DSQuotaExceededException(getSpaceQuota(), getSpaceConsumed());
- }
- }
-
- @Override
- public String toString() {
- String nsQuota = String.valueOf(getQuota());
- String nsCount = String.valueOf(getFileAndDirectoryCount());
- if (getQuota() == HdfsConstants.QUOTA_DONT_SET) {
- nsQuota = "-";
- nsCount = "-";
- }
-
- String ssQuota = StringUtils.byteDesc(getSpaceQuota());
- String ssCount = StringUtils.byteDesc(getSpaceConsumed());
- if (getSpaceQuota() == HdfsConstants.QUOTA_DONT_SET) {
- ssQuota = "-";
- ssCount = "-";
- }
-
- StringBuilder str = new StringBuilder();
- str.append("[NsQuota: ").append(nsQuota).append("/")
- .append(nsCount);
- str.append(", SsQuota: ").append(ssQuota)
- .append("/").append(ssCount)
- .append("]");
- return str.toString();
- }
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org