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:06:08 UTC
[15/39] hadoop git commit: HDFS-13215. RBF: Move Router to its own
module.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c804b284/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/c804b284/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/c804b284/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/c804b284/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/c804b284/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/c804b284/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();
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c804b284/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcClient.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcClient.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcClient.java
deleted file mode 100644
index d3b7947..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcClient.java
+++ /dev/null
@@ -1,1022 +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.lang.reflect.Constructor;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-import java.net.InetSocketAddress;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.LinkedHashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.TreeMap;
-import java.util.concurrent.Callable;
-import java.util.concurrent.CancellationException;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hdfs.DFSConfigKeys;
-import org.apache.hadoop.hdfs.NameNodeProxiesClient.ProxyAndInfo;
-import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys;
-import org.apache.hadoop.hdfs.protocol.ClientProtocol;
-import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
-import org.apache.hadoop.hdfs.server.federation.resolver.ActiveNamenodeResolver;
-import org.apache.hadoop.hdfs.server.federation.resolver.FederationNamenodeContext;
-import org.apache.hadoop.hdfs.server.federation.resolver.FederationNamenodeServiceState;
-import org.apache.hadoop.hdfs.server.federation.resolver.RemoteLocation;
-import org.apache.hadoop.io.retry.RetryPolicies;
-import org.apache.hadoop.io.retry.RetryPolicy;
-import org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision;
-import org.apache.hadoop.ipc.RemoteException;
-import org.apache.hadoop.ipc.StandbyException;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-
-/**
- * A client proxy for Router -> NN communication using the NN ClientProtocol.
- * <p>
- * Provides routers to invoke remote ClientProtocol methods and handle
- * retries/failover.
- * <ul>
- * <li>invokeSingle Make a single request to a single namespace
- * <li>invokeSequential Make a sequential series of requests to multiple
- * ordered namespaces until a condition is met.
- * <li>invokeConcurrent Make concurrent requests to multiple namespaces and
- * return all of the results.
- * </ul>
- * Also maintains a cached pool of connections to NNs. Connections are managed
- * by the ConnectionManager and are unique to each user + NN. The size of the
- * connection pool can be configured. Larger pools allow for more simultaneous
- * requests to a single NN from a single user.
- */
-public class RouterRpcClient {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(RouterRpcClient.class);
-
-
- /** Router identifier. */
- private final String routerId;
-
- /** Interface to identify the active NN for a nameservice or blockpool ID. */
- private final ActiveNamenodeResolver namenodeResolver;
-
- /** Connection pool to the Namenodes per user for performance. */
- private final ConnectionManager connectionManager;
- /** Service to run asynchronous calls. */
- private final ExecutorService executorService;
- /** Retry policy for router -> NN communication. */
- private final RetryPolicy retryPolicy;
- /** Optional perf monitor. */
- private final RouterRpcMonitor rpcMonitor;
-
- /** Pattern to parse a stack trace line. */
- private static final Pattern STACK_TRACE_PATTERN =
- Pattern.compile("\\tat (.*)\\.(.*)\\((.*):(\\d*)\\)");
-
-
- /**
- * Create a router RPC client to manage remote procedure calls to NNs.
- *
- * @param conf Hdfs Configuation.
- * @param resolver A NN resolver to determine the currently active NN in HA.
- * @param monitor Optional performance monitor.
- */
- public RouterRpcClient(Configuration conf, String identifier,
- ActiveNamenodeResolver resolver, RouterRpcMonitor monitor) {
- this.routerId = identifier;
-
- this.namenodeResolver = resolver;
-
- this.connectionManager = new ConnectionManager(conf);
- this.connectionManager.start();
-
- int numThreads = conf.getInt(
- DFSConfigKeys.DFS_ROUTER_CLIENT_THREADS_SIZE,
- DFSConfigKeys.DFS_ROUTER_CLIENT_THREADS_SIZE_DEFAULT);
- ThreadFactory threadFactory = new ThreadFactoryBuilder()
- .setNameFormat("RPC Router Client-%d")
- .build();
- this.executorService = Executors.newFixedThreadPool(
- numThreads, threadFactory);
-
- this.rpcMonitor = monitor;
-
- int maxFailoverAttempts = conf.getInt(
- HdfsClientConfigKeys.Failover.MAX_ATTEMPTS_KEY,
- HdfsClientConfigKeys.Failover.MAX_ATTEMPTS_DEFAULT);
- int maxRetryAttempts = conf.getInt(
- DFSConfigKeys.DFS_ROUTER_CLIENT_MAX_ATTEMPTS,
- DFSConfigKeys.DFS_ROUTER_CLIENT_MAX_ATTEMPTS_DEFAULT);
- int failoverSleepBaseMillis = conf.getInt(
- HdfsClientConfigKeys.Failover.SLEEPTIME_BASE_KEY,
- HdfsClientConfigKeys.Failover.SLEEPTIME_BASE_DEFAULT);
- int failoverSleepMaxMillis = conf.getInt(
- HdfsClientConfigKeys.Failover.SLEEPTIME_MAX_KEY,
- HdfsClientConfigKeys.Failover.SLEEPTIME_MAX_DEFAULT);
- this.retryPolicy = RetryPolicies.failoverOnNetworkException(
- RetryPolicies.TRY_ONCE_THEN_FAIL, maxFailoverAttempts, maxRetryAttempts,
- failoverSleepBaseMillis, failoverSleepMaxMillis);
- }
-
- /**
- * Get the active namenode resolver used by this client.
- * @return Active namenode resolver.
- */
- public ActiveNamenodeResolver getNamenodeResolver() {
- return this.namenodeResolver;
- }
-
- /**
- * Shutdown the client.
- */
- public void shutdown() {
- if (this.connectionManager != null) {
- this.connectionManager.close();
- }
- if (this.executorService != null) {
- this.executorService.shutdownNow();
- }
- }
-
- /**
- * Total number of available sockets between the router and NNs.
- *
- * @return Number of namenode clients.
- */
- public int getNumConnections() {
- return this.connectionManager.getNumConnections();
- }
-
- /**
- * Total number of available sockets between the router and NNs.
- *
- * @return Number of namenode clients.
- */
- public int getNumActiveConnections() {
- return this.connectionManager.getNumActiveConnections();
- }
-
- /**
- * Total number of open connection pools to a NN. Each connection pool.
- * represents one user + one NN.
- *
- * @return Number of connection pools.
- */
- public int getNumConnectionPools() {
- return this.connectionManager.getNumConnectionPools();
- }
-
- /**
- * Number of connections between the router and NNs being created sockets.
- *
- * @return Number of connections waiting to be created.
- */
- public int getNumCreatingConnections() {
- return this.connectionManager.getNumCreatingConnections();
- }
-
- /**
- * JSON representation of the connection pool.
- *
- * @return String representation of the JSON.
- */
- public String getJSON() {
- return this.connectionManager.getJSON();
- }
-
- /**
- * Get ClientProtocol proxy client for a NameNode. Each combination of user +
- * NN must use a unique proxy client. Previously created clients are cached
- * and stored in a connection pool by the ConnectionManager.
- *
- * @param ugi User group information.
- * @param nsId Nameservice identifier.
- * @param rpcAddress ClientProtocol RPC server address of the NN.
- * @return ConnectionContext containing a ClientProtocol proxy client for the
- * NN + current user.
- * @throws IOException If we cannot get a connection to the NameNode.
- */
- private ConnectionContext getConnection(
- UserGroupInformation ugi, String nsId, String rpcAddress)
- throws IOException {
- ConnectionContext connection = null;
- try {
- // Each proxy holds the UGI info for the current user when it is created.
- // This cache does not scale very well, one entry per user per namenode,
- // and may need to be adjusted and/or selectively pruned. The cache is
- // important due to the excessive overhead of creating a new proxy wrapper
- // for each individual request.
-
- // TODO Add tokens from the federated UGI
- connection = this.connectionManager.getConnection(ugi, rpcAddress);
- LOG.debug("User {} NN {} is using connection {}",
- ugi.getUserName(), rpcAddress, connection);
- } catch (Exception ex) {
- LOG.error("Cannot open NN client to address: {}", rpcAddress, ex);
- }
-
- if (connection == null) {
- throw new IOException("Cannot get a connection to " + rpcAddress);
- }
- return connection;
- }
-
- /**
- * Convert an exception to an IOException.
- *
- * For a non-IOException, wrap it with IOException. For a RemoteException,
- * unwrap it. For an IOException which is not a RemoteException, return it.
- *
- * @param e Exception to convert into an exception.
- * @return Created IO exception.
- */
- private static IOException toIOException(Exception e) {
- if (e instanceof RemoteException) {
- return ((RemoteException) e).unwrapRemoteException();
- }
- if (e instanceof IOException) {
- return (IOException)e;
- }
- return new IOException(e);
- }
-
- /**
- * If we should retry the RPC call.
- *
- * @param ioe IOException reported.
- * @param retryCount Number of retries.
- * @param nsId Nameservice ID.
- * @return Retry decision.
- * @throws IOException Original exception if the retry policy generates one
- * or IOException for no available namenodes.
- */
- private RetryDecision shouldRetry(final IOException ioe, final int retryCount,
- final String nsId) throws IOException {
- // check for the case of cluster unavailable state
- if (isClusterUnAvailable(nsId)) {
- // we allow to retry once if cluster is unavailable
- if (retryCount == 0) {
- return RetryDecision.RETRY;
- } else {
- throw new IOException("No namenode available under nameservice " + nsId,
- ioe);
- }
- }
-
- try {
- final RetryPolicy.RetryAction a =
- this.retryPolicy.shouldRetry(ioe, retryCount, 0, true);
- return a.action;
- } catch (Exception ex) {
- LOG.error("Re-throwing API exception, no more retries", ex);
- throw toIOException(ex);
- }
- }
-
- /**
- * Invokes a method against the ClientProtocol proxy server. If a standby
- * exception is generated by the call to the client, retries using the
- * alternate server.
- *
- * Re-throws exceptions generated by the remote RPC call as either
- * RemoteException or IOException.
- *
- * @param ugi User group information.
- * @param namenodes A prioritized list of namenodes within the same
- * nameservice.
- * @param method Remote ClientProtcol method to invoke.
- * @param params Variable list of parameters matching the method.
- * @return The result of invoking the method.
- * @throws IOException
- */
- private Object invokeMethod(
- final UserGroupInformation ugi,
- final List<? extends FederationNamenodeContext> namenodes,
- final Method method, final Object... params) throws IOException {
-
- if (namenodes == null || namenodes.isEmpty()) {
- throw new IOException("No namenodes to invoke " + method.getName() +
- " with params " + Arrays.toString(params) + " from " + this.routerId);
- }
-
- Object ret = null;
- if (rpcMonitor != null) {
- rpcMonitor.proxyOp();
- }
- boolean failover = false;
- Map<FederationNamenodeContext, IOException> ioes = new LinkedHashMap<>();
- for (FederationNamenodeContext namenode : namenodes) {
- ConnectionContext connection = null;
- try {
- String nsId = namenode.getNameserviceId();
- String rpcAddress = namenode.getRpcAddress();
- connection = this.getConnection(ugi, nsId, rpcAddress);
- ProxyAndInfo<ClientProtocol> client = connection.getClient();
- ClientProtocol proxy = client.getProxy();
- ret = invoke(nsId, 0, method, proxy, params);
- if (failover) {
- // Success on alternate server, update
- InetSocketAddress address = client.getAddress();
- namenodeResolver.updateActiveNamenode(nsId, address);
- }
- if (this.rpcMonitor != null) {
- this.rpcMonitor.proxyOpComplete(true);
- }
- return ret;
- } catch (IOException ioe) {
- ioes.put(namenode, ioe);
- if (ioe instanceof StandbyException) {
- // Fail over indicated by retry policy and/or NN
- if (this.rpcMonitor != null) {
- this.rpcMonitor.proxyOpFailureStandby();
- }
- failover = true;
- } else if (ioe instanceof RemoteException) {
- if (this.rpcMonitor != null) {
- this.rpcMonitor.proxyOpComplete(true);
- }
- // RemoteException returned by NN
- throw (RemoteException) ioe;
- } else {
- // Other communication error, this is a failure
- // Communication retries are handled by the retry policy
- if (this.rpcMonitor != null) {
- this.rpcMonitor.proxyOpFailureCommunicate();
- this.rpcMonitor.proxyOpComplete(false);
- }
- throw ioe;
- }
- } finally {
- if (connection != null) {
- connection.release();
- }
- }
- }
- if (this.rpcMonitor != null) {
- this.rpcMonitor.proxyOpComplete(false);
- }
-
- // All namenodes were unavailable or in standby
- String msg = "No namenode available to invoke " + method.getName() + " " +
- Arrays.toString(params);
- LOG.error(msg);
- for (Entry<FederationNamenodeContext, IOException> entry :
- ioes.entrySet()) {
- FederationNamenodeContext namenode = entry.getKey();
- String nsId = namenode.getNameserviceId();
- String nnId = namenode.getNamenodeId();
- String addr = namenode.getRpcAddress();
- IOException ioe = entry.getValue();
- if (ioe instanceof StandbyException) {
- LOG.error("{} {} at {} is in Standby", nsId, nnId, addr);
- } else {
- LOG.error("{} {} at {} error: \"{}\"",
- nsId, nnId, addr, ioe.getMessage());
- }
- }
- throw new StandbyException(msg);
- }
-
- /**
- * Invokes a method on the designated object. Catches exceptions specific to
- * the invocation.
- *
- * Re-throws exceptions generated by the remote RPC call as either
- * RemoteException or IOException.
- *
- * @param nsId Identifier for the namespace
- * @param retryCount Current retry times
- * @param method Method to invoke
- * @param obj Target object for the method
- * @param params Variable parameters
- * @return Response from the remote server
- * @throws IOException
- * @throws InterruptedException
- */
- private Object invoke(String nsId, int retryCount, final Method method,
- final Object obj, final Object... params) throws IOException {
- try {
- return method.invoke(obj, params);
- } catch (IllegalAccessException e) {
- LOG.error("Unexpected exception while proxying API", e);
- return null;
- } catch (IllegalArgumentException e) {
- LOG.error("Unexpected exception while proxying API", e);
- return null;
- } catch (InvocationTargetException e) {
- Throwable cause = e.getCause();
- if (cause instanceof IOException) {
- IOException ioe = (IOException) cause;
-
- // Check if we should retry.
- RetryDecision decision = shouldRetry(ioe, retryCount, nsId);
- if (decision == RetryDecision.RETRY) {
- if (this.rpcMonitor != null) {
- this.rpcMonitor.proxyOpRetries();
- }
-
- // retry
- return invoke(nsId, ++retryCount, method, obj, params);
- } else if (decision == RetryDecision.FAILOVER_AND_RETRY) {
- // failover, invoker looks for standby exceptions for failover.
- if (ioe instanceof StandbyException) {
- throw ioe;
- } else {
- throw new StandbyException(ioe.getMessage());
- }
- } else {
- if (ioe instanceof RemoteException) {
- RemoteException re = (RemoteException) ioe;
- ioe = re.unwrapRemoteException();
- ioe = getCleanException(ioe);
- }
- throw ioe;
- }
- } else {
- throw new IOException(e);
- }
- }
- }
-
- /**
- * Check if the cluster of given nameservice id is available.
- * @param nsId nameservice ID.
- * @return
- * @throws IOException
- */
- private boolean isClusterUnAvailable(String nsId) throws IOException {
- List<? extends FederationNamenodeContext> nnState = this.namenodeResolver
- .getNamenodesForNameserviceId(nsId);
-
- if (nnState != null) {
- for (FederationNamenodeContext nnContext : nnState) {
- // Once we find one NN is in active state, we assume this
- // cluster is available.
- if (nnContext.getState() == FederationNamenodeServiceState.ACTIVE) {
- return false;
- }
- }
- }
-
- return true;
- }
-
- /**
- * Get a clean copy of the exception. Sometimes the exceptions returned by the
- * server contain the full stack trace in the message.
- *
- * @param ioe Exception to clean up.
- * @return Copy of the original exception with a clean message.
- */
- private static IOException getCleanException(IOException ioe) {
- IOException ret = null;
-
- String msg = ioe.getMessage();
- Throwable cause = ioe.getCause();
- StackTraceElement[] stackTrace = ioe.getStackTrace();
-
- // Clean the message by removing the stack trace
- int index = msg.indexOf("\n");
- if (index > 0) {
- String[] msgSplit = msg.split("\n");
- msg = msgSplit[0];
-
- // Parse stack trace from the message
- List<StackTraceElement> elements = new LinkedList<>();
- for (int i=1; i<msgSplit.length; i++) {
- String line = msgSplit[i];
- Matcher matcher = STACK_TRACE_PATTERN.matcher(line);
- if (matcher.find()) {
- String declaringClass = matcher.group(1);
- String methodName = matcher.group(2);
- String fileName = matcher.group(3);
- int lineNumber = Integer.parseInt(matcher.group(4));
- StackTraceElement element = new StackTraceElement(
- declaringClass, methodName, fileName, lineNumber);
- elements.add(element);
- }
- }
- stackTrace = elements.toArray(new StackTraceElement[elements.size()]);
- }
-
- // Create the new output exception
- if (ioe instanceof RemoteException) {
- RemoteException re = (RemoteException)ioe;
- ret = new RemoteException(re.getClassName(), msg);
- } else {
- // Try the simple constructor and initialize the fields
- Class<? extends IOException> ioeClass = ioe.getClass();
- try {
- Constructor<? extends IOException> constructor =
- ioeClass.getDeclaredConstructor(String.class);
- ret = constructor.newInstance(msg);
- } catch (ReflectiveOperationException e) {
- // If there are errors, just use the input one
- LOG.error("Could not create exception {}", ioeClass.getSimpleName(), e);
- ret = ioe;
- }
- }
- if (ret != null) {
- ret.initCause(cause);
- ret.setStackTrace(stackTrace);
- }
-
- return ret;
- }
-
- /**
- * Invokes a ClientProtocol method. Determines the target nameservice via a
- * provided block.
- *
- * Re-throws exceptions generated by the remote RPC call as either
- * RemoteException or IOException.
- *
- * @param block Block used to determine appropriate nameservice.
- * @param method The remote method and parameters to invoke.
- * @return The result of invoking the method.
- * @throws IOException
- */
- public Object invokeSingle(final ExtendedBlock block, RemoteMethod method)
- throws IOException {
- String bpId = block.getBlockPoolId();
- return invokeSingleBlockPool(bpId, method);
- }
-
- /**
- * Invokes a ClientProtocol method. Determines the target nameservice using
- * the block pool id.
- *
- * Re-throws exceptions generated by the remote RPC call as either
- * RemoteException or IOException.
- *
- * @param bpId Block pool identifier.
- * @param method The remote method and parameters to invoke.
- * @return The result of invoking the method.
- * @throws IOException
- */
- public Object invokeSingleBlockPool(final String bpId, RemoteMethod method)
- throws IOException {
- String nsId = getNameserviceForBlockPoolId(bpId);
- return invokeSingle(nsId, method);
- }
-
- /**
- * Invokes a ClientProtocol method against the specified namespace.
- *
- * Re-throws exceptions generated by the remote RPC call as either
- * RemoteException or IOException.
- *
- * @param nsId Target namespace for the method.
- * @param method The remote method and parameters to invoke.
- * @return The result of invoking the method.
- * @throws IOException
- */
- public Object invokeSingle(final String nsId, RemoteMethod method)
- throws IOException {
- UserGroupInformation ugi = RouterRpcServer.getRemoteUser();
- List<? extends FederationNamenodeContext> nns =
- getNamenodesForNameservice(nsId);
- RemoteLocationContext loc = new RemoteLocation(nsId, "/");
- return invokeMethod(ugi, nns, method.getMethod(), method.getParams(loc));
- }
-
- /**
- * Invokes a single proxy call for a single location.
- *
- * Re-throws exceptions generated by the remote RPC call as either
- * RemoteException or IOException.
- *
- * @param location RemoteLocation to invoke.
- * @param remoteMethod The remote method and parameters to invoke.
- * @return The result of invoking the method if successful.
- * @throws IOException
- */
- public Object invokeSingle(final RemoteLocationContext location,
- RemoteMethod remoteMethod) throws IOException {
- List<RemoteLocationContext> locations = Collections.singletonList(location);
- return invokeSequential(locations, remoteMethod);
- }
-
- /**
- * Invokes sequential proxy calls to different locations. Continues to invoke
- * calls until a call returns without throwing a remote exception.
- *
- * @param locations List of locations/nameservices to call concurrently.
- * @param remoteMethod The remote method and parameters to invoke.
- * @return The result of the first successful call, or if no calls are
- * successful, the result of the last RPC call executed.
- * @throws IOException if the success condition is not met and one of the RPC
- * calls generated a remote exception.
- */
- public Object invokeSequential(
- final List<? extends RemoteLocationContext> locations,
- final RemoteMethod remoteMethod) throws IOException {
- return invokeSequential(locations, remoteMethod, null, null);
- }
-
- /**
- * Invokes sequential proxy calls to different locations. Continues to invoke
- * calls until the success condition is met, or until all locations have been
- * attempted.
- *
- * The success condition may be specified by:
- * <ul>
- * <li>An expected result class
- * <li>An expected result value
- * </ul>
- *
- * If no expected result class/values are specified, the success condition is
- * a call that does not throw a remote exception.
- *
- * @param locations List of locations/nameservices to call concurrently.
- * @param remoteMethod The remote method and parameters to invoke.
- * @param expectedResultClass In order to be considered a positive result, the
- * return type must be of this class.
- * @param expectedResultValue In order to be considered a positive result, the
- * return value must equal the value of this object.
- * @return The result of the first successful call, or if no calls are
- * successful, the result of the first RPC call executed.
- * @throws IOException if the success condition is not met, return the first
- * remote exception generated.
- */
- public <T> T invokeSequential(
- final List<? extends RemoteLocationContext> locations,
- final RemoteMethod remoteMethod, Class<T> expectedResultClass,
- Object expectedResultValue) throws IOException {
-
- final UserGroupInformation ugi = RouterRpcServer.getRemoteUser();
- final Method m = remoteMethod.getMethod();
- IOException firstThrownException = null;
- IOException lastThrownException = null;
- Object firstResult = null;
- // Invoke in priority order
- for (final RemoteLocationContext loc : locations) {
- String ns = loc.getNameserviceId();
- List<? extends FederationNamenodeContext> namenodes =
- getNamenodesForNameservice(ns);
- try {
- Object[] params = remoteMethod.getParams(loc);
- Object result = invokeMethod(ugi, namenodes, m, params);
- // Check if the result is what we expected
- if (isExpectedClass(expectedResultClass, result) &&
- isExpectedValue(expectedResultValue, result)) {
- // Valid result, stop here
- @SuppressWarnings("unchecked")
- T ret = (T)result;
- return ret;
- }
- if (firstResult == null) {
- firstResult = result;
- }
- } catch (IOException ioe) {
- // Record it and move on
- lastThrownException = (IOException) ioe;
- if (firstThrownException == null) {
- firstThrownException = lastThrownException;
- }
- } catch (Exception e) {
- // Unusual error, ClientProtocol calls always use IOException (or
- // RemoteException). Re-wrap in IOException for compatibility with
- // ClientProtcol.
- LOG.error("Unexpected exception {} proxying {} to {}",
- e.getClass(), m.getName(), ns, e);
- lastThrownException = new IOException(
- "Unexpected exception proxying API " + e.getMessage(), e);
- if (firstThrownException == null) {
- firstThrownException = lastThrownException;
- }
- }
- }
-
- if (firstThrownException != null) {
- // re-throw the last exception thrown for compatibility
- throw firstThrownException;
- }
- // Return the last result, whether it is the value we are looking for or a
- @SuppressWarnings("unchecked")
- T ret = (T)firstResult;
- return ret;
- }
-
- /**
- * Checks if a result matches the required result class.
- *
- * @param expectedClass Required result class, null to skip the check.
- * @param clazz The result to check.
- * @return True if the result is an instance of the required class or if the
- * expected class is null.
- */
- private static boolean isExpectedClass(Class<?> expectedClass, Object clazz) {
- if (expectedClass == null) {
- return true;
- } else if (clazz == null) {
- return false;
- } else {
- return expectedClass.isInstance(clazz);
- }
- }
-
- /**
- * Checks if a result matches the expected value.
- *
- * @param expectedValue The expected value, null to skip the check.
- * @param value The result to check.
- * @return True if the result is equals to the expected value or if the
- * expected value is null.
- */
- private static boolean isExpectedValue(Object expectedValue, Object value) {
- if (expectedValue == null) {
- return true;
- } else if (value == null) {
- return false;
- } else {
- return value.equals(expectedValue);
- }
- }
-
- /**
- * Invoke multiple concurrent proxy calls to different clients. Returns an
- * array of results.
- *
- * Re-throws exceptions generated by the remote RPC call as either
- * RemoteException or IOException.
- *
- * @param <T> The type of the remote location.
- * @param locations List of remote locations to call concurrently.
- * @param method The remote method and parameters to invoke.
- * @param requireResponse If true an exception will be thrown if all calls do
- * not complete. If false exceptions are ignored and all data results
- * successfully received are returned.
- * @param standby If the requests should go to the standby namenodes too.
- * @throws IOException If all the calls throw an exception.
- */
- public <T extends RemoteLocationContext, R> void invokeConcurrent(
- final Collection<T> locations, final RemoteMethod method,
- boolean requireResponse, boolean standby) throws IOException {
- invokeConcurrent(locations, method, requireResponse, standby, void.class);
- }
-
- /**
- * Invokes multiple concurrent proxy calls to different clients. Returns an
- * array of results.
- *
- * Re-throws exceptions generated by the remote RPC call as either
- * RemoteException or IOException.
- *
- * @param <T> The type of the remote location.
- * @param <R> The type of the remote method return.
- * @param locations List of remote locations to call concurrently.
- * @param method The remote method and parameters to invoke.
- * @param requireResponse If true an exception will be thrown if all calls do
- * not complete. If false exceptions are ignored and all data results
- * successfully received are returned.
- * @param standby If the requests should go to the standby namenodes too.
- * @param clazz Type of the remote return type.
- * @return Result of invoking the method per subcluster: nsId -> result.
- * @throws IOException If requiredResponse=true and any of the calls throw an
- * exception.
- */
- public <T extends RemoteLocationContext, R> Map<T, R> invokeConcurrent(
- final Collection<T> locations, final RemoteMethod method,
- boolean requireResponse, boolean standby, Class<R> clazz)
- throws IOException {
- return invokeConcurrent(
- locations, method, requireResponse, standby, -1, clazz);
- }
-
- /**
- * Invokes multiple concurrent proxy calls to different clients. Returns an
- * array of results.
- *
- * Re-throws exceptions generated by the remote RPC call as either
- * RemoteException or IOException.
- *
- * @param <T> The type of the remote location.
- * @param <R> The type of the remote method return.
- * @param locations List of remote locations to call concurrently.
- * @param method The remote method and parameters to invoke.
- * @param requireResponse If true an exception will be thrown if all calls do
- * not complete. If false exceptions are ignored and all data results
- * successfully received are returned.
- * @param standby If the requests should go to the standby namenodes too.
- * @param timeOutMs Timeout for each individual call.
- * @param clazz Type of the remote return type.
- * @return Result of invoking the method per subcluster: nsId -> result.
- * @throws IOException If requiredResponse=true and any of the calls throw an
- * exception.
- */
- @SuppressWarnings("unchecked")
- public <T extends RemoteLocationContext, R> Map<T, R> invokeConcurrent(
- final Collection<T> locations, final RemoteMethod method,
- boolean requireResponse, boolean standby, long timeOutMs, Class<R> clazz)
- throws IOException {
-
- final UserGroupInformation ugi = RouterRpcServer.getRemoteUser();
- final Method m = method.getMethod();
-
- if (locations.size() == 1) {
- // Shortcut, just one call
- T location = locations.iterator().next();
- String ns = location.getNameserviceId();
- final List<? extends FederationNamenodeContext> namenodes =
- getNamenodesForNameservice(ns);
- Object[] paramList = method.getParams(location);
- Object result = invokeMethod(ugi, namenodes, m, paramList);
- return Collections.singletonMap(location, clazz.cast(result));
- }
-
- List<T> orderedLocations = new LinkedList<>();
- Set<Callable<Object>> callables = new HashSet<>();
- for (final T location : locations) {
- String nsId = location.getNameserviceId();
- final List<? extends FederationNamenodeContext> namenodes =
- getNamenodesForNameservice(nsId);
- final Object[] paramList = method.getParams(location);
- if (standby) {
- // Call the objectGetter to all NNs (including standby)
- for (final FederationNamenodeContext nn : namenodes) {
- String nnId = nn.getNamenodeId();
- final List<FederationNamenodeContext> nnList =
- Collections.singletonList(nn);
- T nnLocation = location;
- if (location instanceof RemoteLocation) {
- nnLocation = (T)new RemoteLocation(nsId, nnId, location.getDest());
- }
- orderedLocations.add(nnLocation);
- callables.add(new Callable<Object>() {
- public Object call() throws Exception {
- return invokeMethod(ugi, nnList, m, paramList);
- }
- });
- }
- } else {
- // Call the objectGetter in order of nameservices in the NS list
- orderedLocations.add(location);
- callables.add(new Callable<Object>() {
- public Object call() throws Exception {
- return invokeMethod(ugi, namenodes, m, paramList);
- }
- });
- }
- }
-
- if (rpcMonitor != null) {
- rpcMonitor.proxyOp();
- }
-
- try {
- List<Future<Object>> futures = null;
- if (timeOutMs > 0) {
- futures = executorService.invokeAll(
- callables, timeOutMs, TimeUnit.MILLISECONDS);
- } else {
- futures = executorService.invokeAll(callables);
- }
- Map<T, R> results = new TreeMap<>();
- Map<T, IOException> exceptions = new TreeMap<>();
- for (int i=0; i<futures.size(); i++) {
- T location = orderedLocations.get(i);
- try {
- Future<Object> future = futures.get(i);
- Object result = future.get();
- results.put(location, clazz.cast(result));
- } catch (CancellationException ce) {
- T loc = orderedLocations.get(i);
- String msg =
- "Invocation to \"" + loc + "\" for \"" + method + "\" timed out";
- LOG.error(msg);
- IOException ioe = new IOException(msg);
- exceptions.put(location, ioe);
- } catch (ExecutionException ex) {
- Throwable cause = ex.getCause();
- LOG.debug("Canot execute {} in {}: {}",
- m.getName(), location, cause.getMessage());
-
- // Convert into IOException if needed
- IOException ioe = null;
- if (cause instanceof IOException) {
- ioe = (IOException) cause;
- } else {
- ioe = new IOException("Unhandled exception while proxying API " +
- m.getName() + ": " + cause.getMessage(), cause);
- }
-
- // Response from all servers required, use this error.
- if (requireResponse) {
- throw ioe;
- }
-
- // Store the exceptions
- exceptions.put(location, ioe);
- }
- }
-
- // Throw the exception for the first location if there are no results
- if (results.isEmpty()) {
- T location = orderedLocations.get(0);
- IOException ioe = exceptions.get(location);
- if (ioe != null) {
- throw ioe;
- }
- }
-
- return results;
- } catch (InterruptedException ex) {
- LOG.error("Unexpected error while invoking API: {}", ex.getMessage());
- throw new IOException(
- "Unexpected error while invoking API " + ex.getMessage(), ex);
- }
- }
-
- /**
- * Get a prioritized list of NNs that share the same nameservice ID (in the
- * same namespace). NNs that are reported as ACTIVE will be first in the list.
- *
- * @param nsId The nameservice ID for the namespace.
- * @return A prioritized list of NNs to use for communication.
- * @throws IOException If a NN cannot be located for the nameservice ID.
- */
- private List<? extends FederationNamenodeContext> getNamenodesForNameservice(
- final String nsId) throws IOException {
-
- final List<? extends FederationNamenodeContext> namenodes =
- namenodeResolver.getNamenodesForNameserviceId(nsId);
-
- if (namenodes == null || namenodes.isEmpty()) {
- throw new IOException("Cannot locate a registered namenode for " + nsId +
- " from " + this.routerId);
- }
- return namenodes;
- }
-
- /**
- * Get a prioritized list of NNs that share the same block pool ID (in the
- * same namespace). NNs that are reported as ACTIVE will be first in the list.
- *
- * @param bpId The blockpool ID for the namespace.
- * @return A prioritized list of NNs to use for communication.
- * @throws IOException If a NN cannot be located for the block pool ID.
- */
- private List<? extends FederationNamenodeContext> getNamenodesForBlockPoolId(
- final String bpId) throws IOException {
-
- List<? extends FederationNamenodeContext> namenodes =
- namenodeResolver.getNamenodesForBlockPoolId(bpId);
-
- if (namenodes == null || namenodes.isEmpty()) {
- throw new IOException("Cannot locate a registered namenode for " + bpId +
- " from " + this.routerId);
- }
- return namenodes;
- }
-
- /**
- * Get the nameservice identifier for a block pool.
- *
- * @param bpId Identifier of the block pool.
- * @return Nameservice identifier.
- * @throws IOException If a NN cannot be located for the block pool ID.
- */
- private String getNameserviceForBlockPoolId(final String bpId)
- throws IOException {
- List<? extends FederationNamenodeContext> namenodes =
- getNamenodesForBlockPoolId(bpId);
- FederationNamenodeContext namenode = namenodes.get(0);
- return namenode.getNameserviceId();
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c804b284/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcMonitor.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcMonitor.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcMonitor.java
deleted file mode 100644
index df9aa11..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcMonitor.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.FederationRPCMetrics;
-import org.apache.hadoop.hdfs.server.federation.store.StateStoreService;
-
-/**
- * Metrics and monitoring interface for the router RPC server. Allows pluggable
- * diagnostics and monitoring services to be attached.
- */
-public interface RouterRpcMonitor {
-
- /**
- * Initialize the monitor.
- * @param conf Configuration for the monitor.
- * @param server RPC server.
- * @param store State Store.
- */
- void init(
- Configuration conf, RouterRpcServer server, StateStoreService store);
-
- /**
- * Get Router RPC metrics info.
- * @return The instance of FederationRPCMetrics.
- */
- FederationRPCMetrics getRPCMetrics();
-
- /**
- * Close the monitor.
- */
- void close();
-
- /**
- * Start processing an operation on the Router.
- */
- void startOp();
-
- /**
- * Start proxying an operation to the Namenode.
- * @return Id of the thread doing the proxying.
- */
- long proxyOp();
-
- /**
- * Mark a proxy operation as completed.
- * @param success If the operation was successful.
- */
- void proxyOpComplete(boolean success);
-
- /**
- * Failed to proxy an operation to a Namenode because it was in standby.
- */
- void proxyOpFailureStandby();
-
- /**
- * Failed to proxy an operation to a Namenode because of an unexpected
- * exception.
- */
- void proxyOpFailureCommunicate();
-
- /**
- * Failed to proxy an operation because it is not implemented.
- */
- void proxyOpNotImplemented();
-
- /**
- * Retry to proxy an operation to a Namenode because of an unexpected
- * exception.
- */
- void proxyOpRetries();
-
- /**
- * If the Router cannot contact the State Store in an operation.
- */
- void routerFailureStateStore();
-
- /**
- * If the Router is in safe mode.
- */
- void routerFailureSafemode();
-
- /**
- * If a path is locked.
- */
- void routerFailureLocked();
-
- /**
- * If a path is in a read only mount point.
- */
- void routerFailureReadOnly();
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org