You are viewing a plain text version of this content. The canonical link for it is here.
Posted to hdfs-commits@hadoop.apache.org by at...@apache.org on 2012/03/03 01:43:00 UTC
svn commit: r1296534 [7/11] - in
/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs: ./ dev-support/
src/contrib/bkjournal/src/main/java/org/apache/hadoop/contrib/bkjournal/
src/contrib/bkjournal/src/test/java/org/apache/hadoop/contrib/bkjournal/
src...
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/JournalSet.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/JournalSet.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/JournalSet.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/JournalSet.java Sat Mar 3 00:42:49 2012
@@ -25,8 +25,10 @@ import java.util.SortedSet;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.hdfs.server.protocol.RemoteEditLog;
import org.apache.hadoop.hdfs.server.protocol.RemoteEditLogManifest;
+
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
@@ -35,8 +37,6 @@ import com.google.common.collect.Lists;
import com.google.common.collect.Multimaps;
import com.google.common.collect.Sets;
-import org.apache.hadoop.classification.InterfaceAudience;
-
/**
* Manages a collection of Journals. None of the methods are synchronized, it is
* assumed that FSEditLog methods, that use this class, use proper
@@ -148,11 +148,17 @@ public class JournalSet implements Journ
private List<JournalAndStream> journals = Lists.newArrayList();
final int minimumRedundantJournals;
+ private volatile Runtime runtime = Runtime.getRuntime();
JournalSet(int minimumRedundantResources) {
this.minimumRedundantJournals = minimumRedundantResources;
}
+ @VisibleForTesting
+ public void setRuntimeForTesting(Runtime runtime) {
+ this.runtime = runtime;
+ }
+
@Override
public EditLogOutputStream startLogSegment(final long txId) throws IOException {
mapJournalsAndReportErrors(new JournalClosure() {
@@ -201,19 +207,25 @@ public class JournalSet implements Journ
* or null if no more exist
*/
@Override
- public EditLogInputStream getInputStream(long fromTxnId) throws IOException {
+ public EditLogInputStream getInputStream(long fromTxnId, boolean inProgressOk)
+ throws IOException {
JournalManager bestjm = null;
long bestjmNumTxns = 0;
CorruptionException corruption = null;
for (JournalAndStream jas : journals) {
+ if (jas.isDisabled()) continue;
+
JournalManager candidate = jas.getManager();
long candidateNumTxns = 0;
try {
- candidateNumTxns = candidate.getNumberOfTransactions(fromTxnId);
+ candidateNumTxns = candidate.getNumberOfTransactions(fromTxnId,
+ inProgressOk);
} catch (CorruptionException ce) {
corruption = ce;
} catch (IOException ioe) {
+ LOG.warn("Unable to read input streams from JournalManager " + candidate,
+ ioe);
continue; // error reading disk, just skip
}
@@ -231,15 +243,20 @@ public class JournalSet implements Journ
return null;
}
}
- return bestjm.getInputStream(fromTxnId);
+ return bestjm.getInputStream(fromTxnId, inProgressOk);
}
@Override
- public long getNumberOfTransactions(long fromTxnId) throws IOException {
+ public long getNumberOfTransactions(long fromTxnId, boolean inProgressOk)
+ throws IOException {
long num = 0;
for (JournalAndStream jas: journals) {
- if (jas.isActive()) {
- long newNum = jas.getManager().getNumberOfTransactions(fromTxnId);
+ if (jas.isDisabled()) {
+ LOG.info("Skipping jas " + jas + " since it's disabled");
+ continue;
+ } else {
+ long newNum = jas.getManager().getNumberOfTransactions(fromTxnId,
+ inProgressOk);
if (newNum > num) {
num = newNum;
}
@@ -298,13 +315,31 @@ public class JournalSet implements Journ
*/
private void mapJournalsAndReportErrors(
JournalClosure closure, String status) throws IOException{
+
List<JournalAndStream> badJAS = Lists.newLinkedList();
for (JournalAndStream jas : journals) {
try {
closure.apply(jas);
} catch (Throwable t) {
- LOG.error("Error: " + status + " failed for (journal " + jas + ")", t);
- badJAS.add(jas);
+ if (jas.isRequired()) {
+ String msg = "Error: " + status + " failed for required journal ("
+ + jas + ")";
+ LOG.fatal(msg, t);
+ // If we fail on *any* of the required journals, then we must not
+ // continue on any of the other journals. Abort them to ensure that
+ // retry behavior doesn't allow them to keep going in any way.
+ abortAllJournals();
+ // the current policy is to shutdown the NN on errors to shared edits
+ // dir. There are many code paths to shared edits failures - syncs,
+ // roll of edits etc. All of them go through this common function
+ // where the isRequired() check is made. Applying exit policy here
+ // to catch all code paths.
+ runtime.exit(1);
+ throw new IOException(msg);
+ } else {
+ LOG.error("Error: " + status + " failed for (journal " + jas + ")", t);
+ badJAS.add(jas);
+ }
}
}
disableAndReportErrorOnJournals(badJAS);
@@ -317,6 +352,17 @@ public class JournalSet implements Journ
}
/**
+ * Abort all of the underlying streams.
+ */
+ private void abortAllJournals() {
+ for (JournalAndStream jas : journals) {
+ if (jas.isActive()) {
+ jas.abort();
+ }
+ }
+ }
+
+ /**
* An implementation of EditLogOutputStream that applies a requested method on
* all the journals that are currently active.
*/
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/LeaseManager.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/LeaseManager.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/LeaseManager.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/LeaseManager.java Sat Mar 3 00:42:49 2012
@@ -34,6 +34,10 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.UnresolvedLinkException;
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.hdfs.server.common.HdfsServerConstants;
+import org.apache.hadoop.util.Daemon;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
import static org.apache.hadoop.hdfs.server.common.Util.now;
@@ -82,6 +86,9 @@ public class LeaseManager {
//
private SortedMap<String, Lease> sortedLeasesByPath = new TreeMap<String, Lease>();
+ private Daemon lmthread;
+ private volatile boolean shouldRunMonitor;
+
LeaseManager(FSNamesystem fsnamesystem) {this.fsnamesystem = fsnamesystem;}
Lease getLease(String holder) {
@@ -146,6 +153,9 @@ public class LeaseManager {
Lease lease = getLease(holder);
if (lease != null) {
removeLease(lease, src);
+ } else {
+ LOG.warn("Removing non-existent lease! holder=" + holder +
+ " src=" + src);
}
}
@@ -190,6 +200,15 @@ public class LeaseManager {
}
}
+ /**
+ * Renew all of the currently open leases.
+ */
+ synchronized void renewAllLeases() {
+ for (Lease l : leases.values()) {
+ renewLease(l);
+ }
+ }
+
/************************************************************
* A Lease governs all the locks held by a single client.
* For each client there's a corresponding lease, whose
@@ -296,6 +315,11 @@ public class LeaseManager {
paths.remove(oldpath);
paths.add(newpath);
}
+
+ @VisibleForTesting
+ long getLastUpdate() {
+ return lastUpdate;
+ }
}
synchronized void changeLease(String src, String dst,
@@ -367,18 +391,18 @@ public class LeaseManager {
/** Check leases periodically. */
public void run() {
- for(; fsnamesystem.isRunning(); ) {
- fsnamesystem.writeLock();
+ for(; shouldRunMonitor && fsnamesystem.isRunning(); ) {
try {
- if (!fsnamesystem.isInSafeMode()) {
- checkLeases();
+ fsnamesystem.writeLockInterruptibly();
+ try {
+ if (!fsnamesystem.isInSafeMode()) {
+ checkLeases();
+ }
+ } finally {
+ fsnamesystem.writeUnlock();
}
- } finally {
- fsnamesystem.writeUnlock();
- }
-
-
- try {
+
+
Thread.sleep(HdfsServerConstants.NAMENODE_LEASE_RECHECK_INTERVAL);
} catch(InterruptedException ie) {
if (LOG.isDebugEnabled()) {
@@ -437,4 +461,36 @@ public class LeaseManager {
+ "\n sortedLeasesByPath=" + sortedLeasesByPath
+ "\n}";
}
+
+ void startMonitor() {
+ Preconditions.checkState(lmthread == null,
+ "Lease Monitor already running");
+ shouldRunMonitor = true;
+ lmthread = new Daemon(new Monitor());
+ lmthread.start();
+ }
+
+ void stopMonitor() {
+ if (lmthread != null) {
+ shouldRunMonitor = false;
+ try {
+ lmthread.interrupt();
+ lmthread.join(3000);
+ } catch (InterruptedException ie) {
+ LOG.warn("Encountered exception ", ie);
+ }
+ lmthread = null;
+ }
+ }
+
+ /**
+ * Trigger the currently-running Lease monitor to re-check
+ * its leases immediately. This is for use by unit tests.
+ */
+ @VisibleForTesting
+ void triggerMonitorCheckNow() {
+ Preconditions.checkState(lmthread != null,
+ "Lease monitor is not running");
+ lmthread.interrupt();
+ }
}
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NNStorage.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NNStorage.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NNStorage.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NNStorage.java Sat Mar 3 00:42:49 2012
@@ -28,6 +28,7 @@ import java.net.URI;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
@@ -161,7 +162,8 @@ public class NNStorage extends Storage i
// this may modify the editsDirs, so copy before passing in
setStorageDirectories(imageDirs,
- Lists.newArrayList(editsDirs));
+ Lists.newArrayList(editsDirs),
+ FSNamesystem.getSharedEditsDirs(conf));
}
@Override // Storage
@@ -249,6 +251,16 @@ public class NNStorage extends Storage i
List<StorageDirectory> getRemovedStorageDirs() {
return this.removedStorageDirs;
}
+
+ /**
+ * See {@link NNStorage#setStorageDirectories(Collection, Collection, Collection)}
+ */
+ @VisibleForTesting
+ synchronized void setStorageDirectories(Collection<URI> fsNameDirs,
+ Collection<URI> fsEditsDirs)
+ throws IOException {
+ setStorageDirectories(fsNameDirs, fsEditsDirs, new ArrayList<URI>());
+ }
/**
* Set the storage directories which will be used. This should only ever be
@@ -265,7 +277,8 @@ public class NNStorage extends Storage i
*/
@VisibleForTesting
synchronized void setStorageDirectories(Collection<URI> fsNameDirs,
- Collection<URI> fsEditsDirs)
+ Collection<URI> fsEditsDirs,
+ Collection<URI> sharedEditsDirs)
throws IOException {
this.storageDirs.clear();
this.removedStorageDirs.clear();
@@ -289,7 +302,8 @@ public class NNStorage extends Storage i
if(dirName.getScheme().compareTo(JournalType.FILE.name().toLowerCase())
== 0){
this.addStorageDir(new StorageDirectory(new File(dirName.getPath()),
- dirType));
+ dirType,
+ !sharedEditsDirs.contains(dirName))); // Don't lock the dir if it's shared.
}
}
@@ -301,7 +315,7 @@ public class NNStorage extends Storage i
if(dirName.getScheme().compareTo(JournalType.FILE.name().toLowerCase())
== 0)
this.addStorageDir(new StorageDirectory(new File(dirName.getPath()),
- NameNodeDirType.EDITS));
+ NameNodeDirType.EDITS, !sharedEditsDirs.contains(dirName)));
}
}
@@ -458,7 +472,7 @@ public class NNStorage extends Storage i
/**
* @return the transaction ID of the last checkpoint.
*/
- long getMostRecentCheckpointTxId() {
+ public long getMostRecentCheckpointTxId() {
return mostRecentCheckpointTxId;
}
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NNStorageRetentionManager.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NNStorageRetentionManager.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NNStorageRetentionManager.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NNStorageRetentionManager.java Sat Mar 3 00:42:49 2012
@@ -31,6 +31,7 @@ import org.apache.hadoop.hdfs.server.nam
import org.apache.hadoop.hdfs.server.namenode.FileJournalManager.EditLogFile;
import org.apache.hadoop.hdfs.util.MD5FileUtils;
+import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
@@ -46,6 +47,7 @@ import com.google.common.collect.Sets;
public class NNStorageRetentionManager {
private final int numCheckpointsToRetain;
+ private final long numExtraEditsToRetain;
private static final Log LOG = LogFactory.getLog(
NNStorageRetentionManager.class);
private final NNStorage storage;
@@ -60,6 +62,15 @@ public class NNStorageRetentionManager {
this.numCheckpointsToRetain = conf.getInt(
DFSConfigKeys.DFS_NAMENODE_NUM_CHECKPOINTS_RETAINED_KEY,
DFSConfigKeys.DFS_NAMENODE_NUM_CHECKPOINTS_RETAINED_DEFAULT);
+ this.numExtraEditsToRetain = conf.getLong(
+ DFSConfigKeys.DFS_NAMENODE_NUM_EXTRA_EDITS_RETAINED_KEY,
+ DFSConfigKeys.DFS_NAMENODE_NUM_EXTRA_EDITS_RETAINED_DEFAULT);
+ Preconditions.checkArgument(numCheckpointsToRetain > 0,
+ "Must retain at least one checkpoint");
+ Preconditions.checkArgument(numExtraEditsToRetain >= 0,
+ DFSConfigKeys.DFS_NAMENODE_NUM_EXTRA_EDITS_RETAINED_KEY +
+ " must not be negative");
+
this.storage = storage;
this.editLog = editLog;
this.purger = purger;
@@ -79,8 +90,12 @@ public class NNStorageRetentionManager {
purgeCheckpointsOlderThan(inspector, minImageTxId);
// If fsimage_N is the image we want to keep, then we need to keep
// all txns > N. We can remove anything < N+1, since fsimage_N
- // reflects the state up to and including N.
- editLog.purgeLogsOlderThan(minImageTxId + 1);
+ // reflects the state up to and including N. However, we also
+ // provide a "cushion" of older txns that we keep, which is
+ // handy for HA, where a remote node may not have as many
+ // new images.
+ long purgeLogsFrom = Math.max(0, minImageTxId + 1 - numExtraEditsToRetain);
+ editLog.purgeLogsOlderThan(purgeLogsFrom);
}
private void purgeCheckpointsOlderThan(
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java Sat Mar 3 00:42:49 2012
@@ -17,9 +17,6 @@
*/
package org.apache.hadoop.hdfs.server.namenode;
-import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY;
-import static org.apache.hadoop.hdfs.DFSConfigKeys.*;
-
import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
@@ -33,22 +30,40 @@ import org.apache.commons.logging.LogFac
import org.apache.hadoop.HadoopIllegalArgumentException;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
+import org.apache.hadoop.ha.HealthCheckFailedException;
+import org.apache.hadoop.ha.ServiceFailedException;
import org.apache.hadoop.fs.CommonConfigurationKeys;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Trash;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.*;
import org.apache.hadoop.hdfs.DFSUtil;
+import org.apache.hadoop.hdfs.HAUtil;
import org.apache.hadoop.hdfs.HdfsConfiguration;
+import org.apache.hadoop.hdfs.protocol.ClientProtocol;
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.NamenodeRole;
import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.StartupOption;
+import org.apache.hadoop.hdfs.server.namenode.ha.ActiveState;
+import org.apache.hadoop.hdfs.server.namenode.ha.HAContext;
+import org.apache.hadoop.hdfs.server.namenode.ha.HAState;
+import org.apache.hadoop.hdfs.server.namenode.ha.StandbyState;
import org.apache.hadoop.hdfs.server.namenode.metrics.NameNodeMetrics;
+import org.apache.hadoop.hdfs.server.protocol.DatanodeProtocol;
+import org.apache.hadoop.hdfs.server.protocol.JournalProtocol;
+import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocol;
import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocols;
import org.apache.hadoop.hdfs.server.protocol.NamenodeRegistration;
+import org.apache.hadoop.ipc.StandbyException;
import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.security.RefreshUserMappingsProtocol;
import org.apache.hadoop.security.SecurityUtil;
import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.authorize.RefreshAuthorizationPolicyProtocol;
+import org.apache.hadoop.tools.GetUserMappingsProtocol;
import org.apache.hadoop.util.ServicePlugin;
import org.apache.hadoop.util.StringUtils;
@@ -96,6 +111,22 @@ public class NameNode {
}
/**
+ * Categories of operations supported by the namenode.
+ */
+ public static enum OperationCategory {
+ /** Operations that are state agnostic */
+ UNCHECKED,
+ /** Read operation that does not change the namespace state */
+ READ,
+ /** Write operation that changes the namespace state */
+ WRITE,
+ /** Operations related to checkpointing */
+ CHECKPOINT,
+ /** Operations related to {@link JournalProtocol} */
+ JOURNAL
+ }
+
+ /**
* HDFS federation configuration can have two types of parameters:
* <ol>
* <li>Parameter that is common for all the name services in the cluster.</li>
@@ -110,6 +141,7 @@ public class NameNode {
DFS_NAMENODE_RPC_ADDRESS_KEY,
DFS_NAMENODE_NAME_DIR_KEY,
DFS_NAMENODE_EDITS_DIR_KEY,
+ DFS_NAMENODE_SHARED_EDITS_DIR_KEY,
DFS_NAMENODE_CHECKPOINT_DIR_KEY,
DFS_NAMENODE_CHECKPOINT_EDITS_DIR_KEY,
DFS_NAMENODE_SERVICE_RPC_ADDRESS_KEY,
@@ -124,14 +156,40 @@ public class NameNode {
DFS_NAMENODE_BACKUP_SERVICE_RPC_ADDRESS_KEY
};
-
+ public long getProtocolVersion(String protocol,
+ long clientVersion) throws IOException {
+ if (protocol.equals(ClientProtocol.class.getName())) {
+ return ClientProtocol.versionID;
+ } else if (protocol.equals(DatanodeProtocol.class.getName())){
+ return DatanodeProtocol.versionID;
+ } else if (protocol.equals(NamenodeProtocol.class.getName())){
+ return NamenodeProtocol.versionID;
+ } else if (protocol.equals(RefreshAuthorizationPolicyProtocol.class.getName())){
+ return RefreshAuthorizationPolicyProtocol.versionID;
+ } else if (protocol.equals(RefreshUserMappingsProtocol.class.getName())){
+ return RefreshUserMappingsProtocol.versionID;
+ } else if (protocol.equals(GetUserMappingsProtocol.class.getName())){
+ return GetUserMappingsProtocol.versionID;
+ } else {
+ throw new IOException("Unknown protocol to name node: " + protocol);
+ }
+ }
+
public static final int DEFAULT_PORT = 8020;
-
public static final Log LOG = LogFactory.getLog(NameNode.class.getName());
public static final Log stateChangeLog = LogFactory.getLog("org.apache.hadoop.hdfs.StateChange");
+ public static final HAState ACTIVE_STATE = new ActiveState();
+ public static final HAState STANDBY_STATE = new StandbyState();
protected FSNamesystem namesystem;
+ protected final Configuration conf;
protected NamenodeRole role;
+ private HAState state;
+ private final boolean haEnabled;
+ private final HAContext haContext;
+ protected boolean allowStaleStandbyReads;
+
+
/** httpServer */
protected NameNodeHttpServer httpServer;
private Thread emptier;
@@ -212,7 +270,7 @@ public class NameNode {
* @param filesystemURI
* @return address of file system
*/
- static InetSocketAddress getAddress(URI filesystemURI) {
+ public static InetSocketAddress getAddress(URI filesystemURI) {
String authority = filesystemURI.getAuthority();
if (authority == null) {
throw new IllegalArgumentException(String.format(
@@ -251,13 +309,11 @@ public class NameNode {
* Given a configuration get the address of the service rpc server
* If the service rpc is not configured returns null
*/
- protected InetSocketAddress getServiceRpcServerAddress(Configuration conf)
- throws IOException {
+ protected InetSocketAddress getServiceRpcServerAddress(Configuration conf) {
return NameNode.getServiceAddress(conf, false);
}
- protected InetSocketAddress getRpcServerAddress(Configuration conf)
- throws IOException {
+ protected InetSocketAddress getRpcServerAddress(Configuration conf) {
return getAddress(conf);
}
@@ -335,11 +391,7 @@ public class NameNode {
throw e;
}
- activate(conf);
- LOG.info(getRole() + " up at: " + rpcServer.getRpcAddress());
- if (rpcServer.getServiceRpcAddress() != null) {
- LOG.info(getRole() + " service server is up at: " + rpcServer.getServiceRpcAddress());
- }
+ startCommonServices(conf);
}
/**
@@ -373,19 +425,11 @@ public class NameNode {
}
}
- /**
- * Activate name-node servers and threads.
- */
- void activate(Configuration conf) throws IOException {
- if ((isRole(NamenodeRole.NAMENODE))
- && (UserGroupInformation.isSecurityEnabled())) {
- namesystem.activateSecretManager();
- }
- namesystem.activate(conf);
+ /** Start the services common to active and standby states */
+ private void startCommonServices(Configuration conf) throws IOException {
+ namesystem.startCommonServices(conf, haContext);
startHttpServer(conf);
rpcServer.start();
- startTrashEmptier(conf);
-
plugins = conf.getInstances(DFS_NAMENODE_PLUGINS_KEY,
ServicePlugin.class);
for (ServicePlugin p: plugins) {
@@ -395,8 +439,28 @@ public class NameNode {
LOG.warn("ServicePlugin " + p + " could not be started", t);
}
}
+ LOG.info(getRole() + " up at: " + rpcServer.getRpcAddress());
+ if (rpcServer.getServiceRpcAddress() != null) {
+ LOG.info(getRole() + " service server is up at: "
+ + rpcServer.getServiceRpcAddress());
+ }
}
-
+
+ private void stopCommonServices() {
+ if(namesystem != null) namesystem.close();
+ if(rpcServer != null) rpcServer.stop();
+ if (plugins != null) {
+ for (ServicePlugin p : plugins) {
+ try {
+ p.stop();
+ } catch (Throwable t) {
+ LOG.warn("ServicePlugin " + p + " could not be stopped", t);
+ }
+ }
+ }
+ stopHttpServer();
+ }
+
private void startTrashEmptier(Configuration conf) throws IOException {
long trashInterval
= conf.getLong(CommonConfigurationKeys.FS_TRASH_INTERVAL_KEY,
@@ -408,11 +472,26 @@ public class NameNode {
this.emptier.start();
}
+ private void stopTrashEmptier() {
+ if (this.emptier != null) {
+ emptier.interrupt();
+ emptier = null;
+ }
+ }
+
private void startHttpServer(final Configuration conf) throws IOException {
httpServer = new NameNodeHttpServer(conf, this, getHttpServerAddress(conf));
httpServer.start();
setHttpServerAddress(conf);
}
+
+ private void stopHttpServer() {
+ try {
+ if (httpServer != null) httpServer.stop();
+ } catch (Exception e) {
+ LOG.error("Exception while stopping httpserver", e);
+ }
+ }
/**
* Start NameNode.
@@ -447,10 +526,23 @@ public class NameNode {
protected NameNode(Configuration conf, NamenodeRole role)
throws IOException {
+ this.conf = conf;
this.role = role;
+ String nsId = getNameServiceId(conf);
+ String namenodeId = HAUtil.getNameNodeId(conf, nsId);
+ this.haEnabled = HAUtil.isHAEnabled(conf, nsId);
+ if (!haEnabled) {
+ state = ACTIVE_STATE;
+ } else {
+ state = STANDBY_STATE;
+ }
+ this.allowStaleStandbyReads = HAUtil.shouldAllowStandbyReads(conf);
+ this.haContext = createHAContext();
try {
- initializeGenericKeys(conf, getNameServiceId(conf));
+ initializeGenericKeys(conf, nsId, namenodeId);
initialize(conf);
+ state.prepareToEnterState(haContext);
+ state.enterState(haContext);
} catch (IOException e) {
this.stop();
throw e;
@@ -460,6 +552,10 @@ public class NameNode {
}
}
+ protected HAContext createHAContext() {
+ return new NameNodeHAContext();
+ }
+
/**
* Wait for service to finish.
* (Normally, it runs forever.)
@@ -468,6 +564,7 @@ public class NameNode {
try {
this.rpcServer.join();
} catch (InterruptedException ie) {
+ LOG.info("Caught interrupted exception ", ie);
}
}
@@ -480,23 +577,14 @@ public class NameNode {
return;
stopRequested = true;
}
- if (plugins != null) {
- for (ServicePlugin p : plugins) {
- try {
- p.stop();
- } catch (Throwable t) {
- LOG.warn("ServicePlugin " + p + " could not be stopped", t);
- }
- }
- }
try {
- if (httpServer != null) httpServer.stop();
- } catch (Exception e) {
- LOG.error("Exception while stopping httpserver", e);
+ if (state != null) {
+ state.exitState(haContext);
+ }
+ } catch (ServiceFailedException e) {
+ LOG.warn("Encountered exception while exiting state ", e);
}
- if(namesystem != null) namesystem.close();
- if(emptier != null) emptier.interrupt();
- if(rpcServer != null) rpcServer.stop();
+ stopCommonServices();
if (metrics != null) {
metrics.shutdown();
}
@@ -561,6 +649,10 @@ public class NameNode {
private static boolean format(Configuration conf,
boolean isConfirmationNeeded)
throws IOException {
+ String nsId = DFSUtil.getNamenodeNameServiceId(conf);
+ String namenodeId = HAUtil.getNameNodeId(conf, nsId);
+ initializeGenericKeys(conf, nsId, namenodeId);
+
if (!conf.getBoolean(DFS_NAMENODE_SUPPORT_ALLOW_FORMAT_KEY,
DFS_NAMENODE_SUPPORT_ALLOW_FORMAT_DEFAULT)) {
throw new IOException("The option " + DFS_NAMENODE_SUPPORT_ALLOW_FORMAT_KEY
@@ -571,7 +663,7 @@ public class NameNode {
}
Collection<URI> dirsToFormat = FSNamesystem.getNamespaceDirs(conf);
- Collection<URI> editDirsToFormat =
+ List<URI> editDirsToFormat =
FSNamesystem.getNamespaceEditsDirs(conf);
for(Iterator<URI> it = dirsToFormat.iterator(); it.hasNext();) {
File curDir = new File(it.next().getPath());
@@ -605,6 +697,10 @@ public class NameNode {
private static boolean finalize(Configuration conf,
boolean isConfirmationNeeded
) throws IOException {
+ String nsId = DFSUtil.getNamenodeNameServiceId(conf);
+ String namenodeId = HAUtil.getNameNodeId(conf, nsId);
+ initializeGenericKeys(conf, nsId, namenodeId);
+
FSNamesystem nsys = new FSNamesystem(conf, new FSImage(conf));
System.err.print(
"\"finalize\" will remove the previous state of the files system.\n"
@@ -721,6 +817,14 @@ public class NameNode {
return null;
}
setStartupOption(conf, startOpt);
+
+ if (HAUtil.isHAEnabled(conf, DFSUtil.getNamenodeNameServiceId(conf)) &&
+ (startOpt == StartupOption.UPGRADE ||
+ startOpt == StartupOption.ROLLBACK ||
+ startOpt == StartupOption.FINALIZE)) {
+ throw new HadoopIllegalArgumentException("Invalid startup option. " +
+ "Cannot perform DFS upgrade with HA enabled.");
+ }
switch (startOpt) {
case FORMAT:
@@ -761,16 +865,26 @@ public class NameNode {
* @param conf
* Configuration object to lookup specific key and to set the value
* to the key passed. Note the conf object is modified
- * @param nameserviceId name service Id
- * @see DFSUtil#setGenericConf(Configuration, String, String...)
- */
- public static void initializeGenericKeys(Configuration conf, String
- nameserviceId) {
- if ((nameserviceId == null) || nameserviceId.isEmpty()) {
+ * @param nameserviceId name service Id (to distinguish federated NNs)
+ * @param namenodeId the namenode ID (to distinguish HA NNs)
+ * @see DFSUtil#setGenericConf(Configuration, String, String, String...)
+ */
+ public static void initializeGenericKeys(Configuration conf,
+ String nameserviceId, String namenodeId) {
+ if ((nameserviceId == null || nameserviceId.isEmpty()) &&
+ (namenodeId == null || namenodeId.isEmpty())) {
return;
}
- DFSUtil.setGenericConf(conf, nameserviceId, NAMESERVICE_SPECIFIC_KEYS);
+ if (nameserviceId != null) {
+ conf.set(DFS_FEDERATION_NAMESERVICE_ID, nameserviceId);
+ }
+ if (namenodeId != null) {
+ conf.set(DFS_HA_NAMENODE_ID_KEY, namenodeId);
+ }
+
+ DFSUtil.setGenericConf(conf, nameserviceId, namenodeId,
+ NAMESERVICE_SPECIFIC_KEYS);
if (conf.get(DFS_NAMENODE_RPC_ADDRESS_KEY) != null) {
URI defaultUri = URI.create(HdfsConstants.HDFS_URI_SCHEME + "://"
+ conf.get(DFS_NAMENODE_RPC_ADDRESS_KEY));
@@ -799,4 +913,131 @@ public class NameNode {
System.exit(-1);
}
}
+
+ synchronized void monitorHealth()
+ throws HealthCheckFailedException, AccessControlException {
+ namesystem.checkSuperuserPrivilege();
+ if (!haEnabled) {
+ return; // no-op, if HA is not enabled
+ }
+ getNamesystem().checkAvailableResources();
+ if (!getNamesystem().nameNodeHasResourcesAvailable()) {
+ throw new HealthCheckFailedException(
+ "The NameNode has no resources available");
+ }
+ }
+
+ synchronized void transitionToActive()
+ throws ServiceFailedException, AccessControlException {
+ namesystem.checkSuperuserPrivilege();
+ if (!haEnabled) {
+ throw new ServiceFailedException("HA for namenode is not enabled");
+ }
+ state.setState(haContext, ACTIVE_STATE);
+ }
+
+ synchronized void transitionToStandby()
+ throws ServiceFailedException, AccessControlException {
+ namesystem.checkSuperuserPrivilege();
+ if (!haEnabled) {
+ throw new ServiceFailedException("HA for namenode is not enabled");
+ }
+ state.setState(haContext, STANDBY_STATE);
+ }
+
+ synchronized HAServiceState getServiceState() throws AccessControlException {
+ namesystem.checkSuperuserPrivilege();
+ if (state == null) {
+ return HAServiceState.INITIALIZING;
+ }
+ return state.getServiceState();
+ }
+
+ synchronized boolean readyToBecomeActive()
+ throws ServiceFailedException, AccessControlException {
+ namesystem.checkSuperuserPrivilege();
+ if (!haEnabled) {
+ throw new ServiceFailedException("HA for namenode is not enabled");
+ }
+ return !isInSafeMode();
+ }
+
+
+ /**
+ * Class used as expose {@link NameNode} as context to {@link HAState}
+ *
+ * TODO(HA):
+ * When entering and exiting state, on failing to start services,
+ * appropriate action is needed todo either shutdown the node or recover
+ * from failure.
+ */
+ protected class NameNodeHAContext implements HAContext {
+ @Override
+ public void setState(HAState s) {
+ state = s;
+ }
+
+ @Override
+ public HAState getState() {
+ return state;
+ }
+
+ @Override
+ public void startActiveServices() throws IOException {
+ namesystem.startActiveServices();
+ startTrashEmptier(conf);
+ }
+
+ @Override
+ public void stopActiveServices() throws IOException {
+ if (namesystem != null) {
+ namesystem.stopActiveServices();
+ }
+ stopTrashEmptier();
+ }
+
+ @Override
+ public void startStandbyServices() throws IOException {
+ namesystem.startStandbyServices();
+ }
+
+ @Override
+ public void prepareToStopStandbyServices() throws ServiceFailedException {
+ namesystem.prepareToStopStandbyServices();
+ }
+
+ @Override
+ public void stopStandbyServices() throws IOException {
+ if (namesystem != null) {
+ namesystem.stopStandbyServices();
+ }
+ }
+
+ @Override
+ public void writeLock() {
+ namesystem.writeLock();
+ }
+
+ @Override
+ public void writeUnlock() {
+ namesystem.writeUnlock();
+ }
+
+ /** Check if an operation of given category is allowed */
+ @Override
+ public void checkOperation(final OperationCategory op)
+ throws StandbyException {
+ state.checkOperation(haContext, op);
+ }
+
+ @Override
+ public boolean allowStaleReads() {
+ return allowStaleStandbyReads;
+ }
+
+ }
+
+ public boolean isStandbyState() {
+ return (state.equals(STANDBY_STATE));
+ }
}
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeResourceChecker.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeResourceChecker.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeResourceChecker.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeResourceChecker.java Sat Mar 3 00:42:49 2012
@@ -46,7 +46,7 @@ import com.google.common.base.Predicate;
* are added by default, and arbitrary extra volumes may be configured as well.
*/
@InterfaceAudience.Private
-class NameNodeResourceChecker {
+public class NameNodeResourceChecker {
private static final Log LOG = LogFactory.getLog(NameNodeResourceChecker.class.getName());
// Space (in bytes) reserved per volume.
@@ -176,8 +176,7 @@ class NameNodeResourceChecker {
* least one redundant volume and all of the required volumes, false
* otherwise.
*/
- boolean hasAvailableDiskSpace()
- throws IOException {
+ public boolean hasAvailableDiskSpace() {
return NameNodeResourcePolicy.areResourcesAvailable(volumes.values(),
minimumRedundantVolumes);
}
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeResourcePolicy.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeResourcePolicy.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeResourcePolicy.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeResourcePolicy.java Sat Mar 3 00:42:49 2012
@@ -37,9 +37,6 @@ final class NameNodeResourcePolicy {
* required to continue operation.
* @return true if and only if there are sufficient NN resources to
* continue logging edits.
- * @throws RuntimeException if the number of <bold>configured</bold>
- * redundant resources is fewer than the minimum number of available
- * redundant resources.
*/
static boolean areResourcesAvailable(
Collection<? extends CheckableNameNodeResource> resources,
@@ -63,12 +60,6 @@ final class NameNodeResourcePolicy {
}
}
- if (redundantResourceCount < minimumRedundantResources) {
- throw new RuntimeException("Need a minimum of " + minimumRedundantResources
- + " for NN to operate but only " + redundantResourceCount
- + " are configured.");
- }
-
if (redundantResourceCount == 0) {
// If there are no redundant resources, return true if there are any
// required resources available.
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java Sat Mar 3 00:42:49 2012
@@ -41,6 +41,13 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.UnresolvedLinkException;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.fs.permission.PermissionStatus;
+import org.apache.hadoop.ha.HAServiceProtocol;
+import org.apache.hadoop.ha.HealthCheckFailedException;
+import org.apache.hadoop.ha.ServiceFailedException;
+import org.apache.hadoop.ha.proto.HAServiceProtocolProtos.HAServiceProtocolService;
+import org.apache.hadoop.ha.protocolPB.HAServiceProtocolPB;
+import org.apache.hadoop.ha.protocolPB.HAServiceProtocolServerSideTranslatorPB;
+
import org.apache.hadoop.hdfs.DFSUtil;
import org.apache.hadoop.hdfs.HDFSPolicyProvider;
import org.apache.hadoop.hdfs.HdfsConfiguration;
@@ -82,6 +89,7 @@ import org.apache.hadoop.hdfs.security.t
import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.NamenodeRole;
import org.apache.hadoop.hdfs.server.common.IncorrectVersionException;
import org.apache.hadoop.hdfs.server.common.UpgradeStatusReport;
+import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory;
import org.apache.hadoop.hdfs.server.namenode.metrics.NameNodeMetrics;
import org.apache.hadoop.hdfs.server.namenode.web.resources.NamenodeWebHdfsMethods;
import org.apache.hadoop.hdfs.server.protocol.BlocksWithLocations;
@@ -90,6 +98,7 @@ import org.apache.hadoop.hdfs.server.pro
import org.apache.hadoop.hdfs.server.protocol.DatanodeRegistration;
import org.apache.hadoop.hdfs.server.protocol.DatanodeStorage;
import org.apache.hadoop.hdfs.server.protocol.FinalizeCommand;
+import org.apache.hadoop.hdfs.server.protocol.HeartbeatResponse;
import org.apache.hadoop.hdfs.server.protocol.NamenodeCommand;
import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocols;
import org.apache.hadoop.hdfs.server.protocol.NamenodeRegistration;
@@ -104,6 +113,7 @@ import org.apache.hadoop.io.EnumSetWrita
import org.apache.hadoop.io.Text;
import org.apache.hadoop.ipc.ProtobufRpcEngine;
import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.ipc.Server;
import org.apache.hadoop.ipc.WritableRpcEngine;
import org.apache.hadoop.net.Node;
@@ -127,7 +137,7 @@ class NameNodeRpcServer implements Namen
private static final Log stateChangeLog = NameNode.stateChangeLog;
// Dependencies from other parts of NN.
- private final FSNamesystem namesystem;
+ protected final FSNamesystem namesystem;
protected final NameNode nn;
private final NameNodeMetrics metrics;
@@ -183,6 +193,11 @@ class NameNodeRpcServer implements Namen
new GetUserMappingsProtocolServerSideTranslatorPB(this);
BlockingService getUserMappingService = GetUserMappingsProtocolService
.newReflectiveBlockingService(getUserMappingXlator);
+
+ HAServiceProtocolServerSideTranslatorPB haServiceProtocolXlator =
+ new HAServiceProtocolServerSideTranslatorPB(this);
+ BlockingService haPbService = HAServiceProtocolService
+ .newReflectiveBlockingService(haServiceProtocolXlator);
WritableRpcEngine.ensureInitialized();
@@ -198,6 +213,8 @@ class NameNodeRpcServer implements Namen
dnSocketAddr.getHostName(), dnSocketAddr.getPort(),
serviceHandlerCount,
false, conf, namesystem.getDelegationTokenSecretManager());
+ DFSUtil.addPBProtocol(conf, HAServiceProtocolPB.class, haPbService,
+ serviceRpcServer);
DFSUtil.addPBProtocol(conf, NamenodeProtocolPB.class, NNPbService,
serviceRpcServer);
DFSUtil.addPBProtocol(conf, DatanodeProtocolPB.class, dnProtoPbService,
@@ -221,6 +238,8 @@ class NameNodeRpcServer implements Namen
clientNNPbService, socAddr.getHostName(),
socAddr.getPort(), handlerCount, false, conf,
namesystem.getDelegationTokenSecretManager());
+ DFSUtil.addPBProtocol(conf, HAServiceProtocolPB.class, haPbService,
+ clientRpcServer);
DFSUtil.addPBProtocol(conf, NamenodeProtocolPB.class, NNPbService,
clientRpcServer);
DFSUtil.addPBProtocol(conf, DatanodeProtocolPB.class, dnProtoPbService,
@@ -287,7 +306,7 @@ class NameNodeRpcServer implements Namen
throw new IllegalArgumentException(
"Unexpected not positive size: "+size);
}
-
+ namesystem.checkOperation(OperationCategory.READ);
return namesystem.getBlockManager().getBlocks(datanode, size);
}
@@ -300,6 +319,7 @@ class NameNodeRpcServer implements Namen
public void errorReport(NamenodeRegistration registration,
int errorCode,
String msg) throws IOException {
+ namesystem.checkOperation(OperationCategory.UNCHECKED);
verifyRequest(registration);
LOG.info("Error report from " + registration + ": " + msg);
if(errorCode == FATAL)
@@ -327,9 +347,6 @@ class NameNodeRpcServer implements Namen
@Override // NamenodeProtocol
public void endCheckpoint(NamenodeRegistration registration,
CheckpointSignature sig) throws IOException {
- verifyRequest(registration);
- if(!nn.isRole(NamenodeRole.NAMENODE))
- throw new IOException("Only an ACTIVE node can invoke endCheckpoint.");
namesystem.endCheckpoint(registration, sig);
}
@@ -478,10 +495,10 @@ class NameNodeRpcServer implements Namen
return namesystem.getAdditionalDatanode(src, blk,
existings, excludeSet, numAdditionalNodes, clientName);
}
-
/**
* The client needs to give up on the block.
*/
+ @Override // ClientProtocol
public void abandonBlock(ExtendedBlock b, String src, String holder)
throws IOException {
if(stateChangeLog.isDebugEnabled()) {
@@ -509,18 +526,9 @@ class NameNodeRpcServer implements Namen
* mark the block as corrupt. In the future we might
* check the blocks are actually corrupt.
*/
- @Override
+ @Override // ClientProtocol, DatanodeProtocol
public void reportBadBlocks(LocatedBlock[] blocks) throws IOException {
- stateChangeLog.info("*DIR* NameNode.reportBadBlocks");
- for (int i = 0; i < blocks.length; i++) {
- ExtendedBlock blk = blocks[i].getBlock();
- DatanodeInfo[] nodes = blocks[i].getLocations();
- for (int j = 0; j < nodes.length; j++) {
- DatanodeInfo dn = nodes[j];
- namesystem.getBlockManager().findAndMarkBlockAsCorrupt(blk, dn,
- "client machine reported it");
- }
- }
+ namesystem.reportBadBlocks(blocks);
}
@Override // ClientProtocol
@@ -633,8 +641,7 @@ class NameNodeRpcServer implements Namen
@Override // ClientProtocol
public DirectoryListing getListing(String src, byte[] startAfter,
- boolean needLocation)
- throws IOException {
+ boolean needLocation) throws IOException {
DirectoryListing files = namesystem.getListing(
src, startAfter, needLocation);
if (files != null) {
@@ -656,14 +663,16 @@ class NameNodeRpcServer implements Namen
return namesystem.getFileInfo(src, false);
}
- @Override
- public long[] getStats() {
+ @Override // ClientProtocol
+ public long[] getStats() throws IOException {
+ namesystem.checkOperation(OperationCategory.READ);
return namesystem.getStats();
}
@Override // ClientProtocol
public DatanodeInfo[] getDatanodeReport(DatanodeReportType type)
- throws IOException {
+ throws IOException {
+ namesystem.checkOperation(OperationCategory.UNCHECKED);
DatanodeInfo results[] = namesystem.datanodeReport(type);
if (results == null ) {
throw new IOException("Cannot find datanode report");
@@ -673,28 +682,32 @@ class NameNodeRpcServer implements Namen
@Override // ClientProtocol
public boolean setSafeMode(SafeModeAction action) throws IOException {
+ namesystem.checkOperation(OperationCategory.UNCHECKED);
return namesystem.setSafeMode(action);
}
@Override // ClientProtocol
- public boolean restoreFailedStorage(String arg)
- throws AccessControlException {
+ public boolean restoreFailedStorage(String arg) throws IOException {
+ namesystem.checkOperation(OperationCategory.UNCHECKED);
return namesystem.restoreFailedStorage(arg);
}
@Override // ClientProtocol
public void saveNamespace() throws IOException {
+ namesystem.checkOperation(OperationCategory.UNCHECKED);
namesystem.saveNamespace();
}
@Override // ClientProtocol
public void refreshNodes() throws IOException {
+ namesystem.checkOperation(OperationCategory.UNCHECKED);
namesystem.getBlockManager().getDatanodeManager().refreshNodes(
new HdfsConfiguration());
}
@Override // NamenodeProtocol
- public long getTransactionID() {
+ public long getTransactionID() throws IOException {
+ namesystem.checkOperation(OperationCategory.CHECKPOINT);
return namesystem.getEditLog().getSyncTxId();
}
@@ -703,32 +716,36 @@ class NameNodeRpcServer implements Namen
return namesystem.rollEditLog();
}
- @Override
+ @Override // NamenodeProtocol
public RemoteEditLogManifest getEditLogManifest(long sinceTxId)
throws IOException {
+ namesystem.checkOperation(OperationCategory.READ);
return namesystem.getEditLog().getEditLogManifest(sinceTxId);
}
@Override // ClientProtocol
public void finalizeUpgrade() throws IOException {
+ namesystem.checkOperation(OperationCategory.WRITE);
namesystem.finalizeUpgrade();
}
@Override // ClientProtocol
public UpgradeStatusReport distributedUpgradeProgress(UpgradeAction action)
throws IOException {
+ namesystem.checkOperation(OperationCategory.READ);
return namesystem.distributedUpgradeProgress(action);
}
@Override // ClientProtocol
public void metaSave(String filename) throws IOException {
+ namesystem.checkOperation(OperationCategory.UNCHECKED);
namesystem.metaSave(filename);
}
@Override // ClientProtocol
public CorruptFileBlocks listCorruptFileBlocks(String path, String cookie)
throws IOException {
- String[] cookieTab = new String[] { cookie };
+ String[] cookieTab = new String[] { cookie };
Collection<FSNamesystem.CorruptFileBlockInfo> fbs =
namesystem.listCorruptFileBlocks(path, cookieTab);
@@ -743,11 +760,12 @@ class NameNodeRpcServer implements Namen
/**
* Tell all datanodes to use a new, non-persistent bandwidth value for
* dfs.datanode.balance.bandwidthPerSec.
- * @param bandwidth Blanacer bandwidth in bytes per second for all datanodes.
+ * @param bandwidth Balancer bandwidth in bytes per second for all datanodes.
* @throws IOException
*/
@Override // ClientProtocol
public void setBalancerBandwidth(long bandwidth) throws IOException {
+ namesystem.checkOperation(OperationCategory.UNCHECKED);
namesystem.getBlockManager().getDatanodeManager().setBalancerBandwidth(bandwidth);
}
@@ -822,7 +840,7 @@ class NameNodeRpcServer implements Namen
}
@Override // DatanodeProtocol
- public DatanodeCommand[] sendHeartbeat(DatanodeRegistration nodeReg,
+ public HeartbeatResponse sendHeartbeat(DatanodeRegistration nodeReg,
StorageReport[] report, int xmitsInProgress, int xceiverCount,
int failedVolumes) throws IOException {
verifyRequest(nodeReg);
@@ -844,7 +862,7 @@ class NameNodeRpcServer implements Namen
}
namesystem.getBlockManager().processReport(nodeReg, poolId, blist);
- if (nn.getFSImage().isUpgradeFinalized())
+ if (nn.getFSImage().isUpgradeFinalized() && !nn.isStandbyState())
return new FinalizeCommand(poolId);
return null;
}
@@ -858,7 +876,7 @@ class NameNodeRpcServer implements Namen
+"from "+nodeReg.getName()+" "+receivedAndDeletedBlocks.length
+" blocks.");
}
- namesystem.getBlockManager().blockReceivedAndDeleted(
+ namesystem.getBlockManager().processIncrementalBlockReport(
nodeReg, poolId, receivedAndDeletedBlocks[0].getBlocks());
}
@@ -946,6 +964,35 @@ class NameNodeRpcServer implements Namen
return UserGroupInformation.createRemoteUser(user).getGroupNames();
}
+ @Override // HAServiceProtocol
+ public synchronized void monitorHealth()
+ throws HealthCheckFailedException, AccessControlException {
+ nn.monitorHealth();
+ }
+
+ @Override // HAServiceProtocol
+ public synchronized void transitionToActive()
+ throws ServiceFailedException, AccessControlException {
+ nn.transitionToActive();
+ }
+
+ @Override // HAServiceProtocol
+ public synchronized void transitionToStandby()
+ throws ServiceFailedException, AccessControlException {
+ nn.transitionToStandby();
+ }
+
+ @Override // HAServiceProtocol
+ public synchronized HAServiceState getServiceState()
+ throws AccessControlException {
+ return nn.getServiceState();
+ }
+
+ @Override // HAServiceProtocol
+ public synchronized boolean readyToBecomeActive()
+ throws ServiceFailedException, AccessControlException {
+ return nn.readyToBecomeActive();
+ }
/**
* Verify version.
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NamenodeJspHelper.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NamenodeJspHelper.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NamenodeJspHelper.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NamenodeJspHelper.java Sat Mar 3 00:42:49 2012
@@ -36,6 +36,7 @@ import javax.servlet.http.HttpServletRes
import javax.servlet.jsp.JspWriter;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DFSUtil;
import org.apache.hadoop.hdfs.protocol.Block;
@@ -309,7 +310,16 @@ class NamenodeJspHelper {
long bpUsed = fsnStats[6];
float percentBpUsed = DFSUtil.getPercentUsed(bpUsed, total);
-
+
+ // don't show under-replicated/missing blocks or corrupt files for SBN
+ // since the standby namenode doesn't compute replication queues
+ String underReplicatedBlocks = "";
+ if (nn.getServiceState() == HAServiceState.ACTIVE) {
+ underReplicatedBlocks = rowTxt()
+ + colTxt("Excludes missing blocks.")
+ + "Number of Under-Replicated Blocks" + colTxt() + ":" + colTxt()
+ + fsn.getBlockManager().getUnderReplicatedNotMissingBlocks();
+ }
out.print("<div id=\"dfstable\"> <table>\n" + rowTxt() + colTxt()
+ "Configured Capacity" + colTxt() + ":" + colTxt()
+ StringUtils.byteDesc(total) + rowTxt() + colTxt() + "DFS Used"
@@ -344,10 +354,8 @@ class NamenodeJspHelper {
+ rowTxt() + colTxt()
+ "<a href=\"dfsnodelist.jsp?whatNodes=DECOMMISSIONING\">"
+ "Decommissioning Nodes</a> "
- + colTxt() + ":" + colTxt() + decommissioning.size()
- + rowTxt() + colTxt("Excludes missing blocks.")
- + "Number of Under-Replicated Blocks" + colTxt() + ":" + colTxt()
- + fsn.getBlockManager().getUnderReplicatedNotMissingBlocks()
+ + colTxt() + ":" + colTxt() + decommissioning.size()
+ + underReplicatedBlocks
+ "</table></div><br>\n");
if (live.isEmpty() && dead.isEmpty()) {
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/Namesystem.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/Namesystem.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/Namesystem.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/Namesystem.java Sat Mar 3 00:42:49 2012
@@ -32,4 +32,10 @@ public interface Namesystem extends RwLo
/** @return the block pool ID */
public String getBlockPoolId();
+
+ public boolean isInStandbyState();
+
+ public boolean isGenStampInFuture(long generationStamp);
+
+ public void adjustSafeModeBlockTotals(int deltaSafe, int deltaTotal);
}
\ No newline at end of file
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/SecondaryNameNode.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/SecondaryNameNode.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/SecondaryNameNode.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/SecondaryNameNode.java Sat Mar 3 00:42:49 2012
@@ -45,10 +45,11 @@ import org.apache.hadoop.fs.FileSystem;
import static org.apache.hadoop.hdfs.DFSConfigKeys.*;
import org.apache.hadoop.hdfs.DFSUtil;
+import org.apache.hadoop.hdfs.HAUtil;
+import org.apache.hadoop.hdfs.NameNodeProxies;
import org.apache.hadoop.hdfs.DFSUtil.ErrorSimulator;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
-import org.apache.hadoop.hdfs.protocolPB.NamenodeProtocolTranslatorPB;
import org.apache.hadoop.hdfs.server.common.HdfsServerConstants;
import org.apache.hadoop.hdfs.server.common.InconsistentFSStateException;
import org.apache.hadoop.hdfs.server.common.JspHelper;
@@ -112,17 +113,9 @@ public class SecondaryNameNode implement
private String infoBindAddress;
private Collection<URI> checkpointDirs;
- private Collection<URI> checkpointEditsDirs;
-
- /** How often to checkpoint regardless of number of txns */
- private long checkpointPeriod; // in seconds
-
- /** How often to poll the NN to check checkpointTxnCount */
- private long checkpointCheckPeriod; // in seconds
-
- /** checkpoint once every this many transactions, regardless of time */
- private long checkpointTxnCount;
+ private List<URI> checkpointEditsDirs;
+ private CheckpointConf checkpointConf;
private FSNamesystem namesystem;
@@ -132,9 +125,9 @@ public class SecondaryNameNode implement
+ "\nName Node Address : " + nameNodeAddr
+ "\nStart Time : " + new Date(starttime)
+ "\nLast Checkpoint Time : " + (lastCheckpointTime == 0? "--": new Date(lastCheckpointTime))
- + "\nCheckpoint Period : " + checkpointPeriod + " seconds"
- + "\nCheckpoint Size : " + StringUtils.byteDesc(checkpointTxnCount)
- + " (= " + checkpointTxnCount + " bytes)"
+ + "\nCheckpoint Period : " + checkpointConf.getPeriod() + " seconds"
+ + "\nCheckpoint Size : " + StringUtils.byteDesc(checkpointConf.getTxnCount())
+ + " (= " + checkpointConf.getTxnCount() + " bytes)"
+ "\nCheckpoint Dirs : " + checkpointDirs
+ "\nCheckpoint Edits Dirs: " + checkpointEditsDirs;
}
@@ -174,16 +167,19 @@ public class SecondaryNameNode implement
public SecondaryNameNode(Configuration conf,
CommandLineOpts commandLineOpts) throws IOException {
try {
- NameNode.initializeGenericKeys(conf,
- DFSUtil.getSecondaryNameServiceId(conf));
+ String nsId = DFSUtil.getSecondaryNameServiceId(conf);
+ if (HAUtil.isHAEnabled(conf, nsId)) {
+ throw new IOException(
+ "Cannot use SecondaryNameNode in an HA cluster." +
+ " The Standby Namenode will perform checkpointing.");
+ }
+ NameNode.initializeGenericKeys(conf, nsId, null);
initialize(conf, commandLineOpts);
- } catch(IOException e) {
+ } catch (IOException e) {
shutdown();
- LOG.fatal("Failed to start secondary namenode. ", e);
throw e;
- } catch(HadoopIllegalArgumentException e) {
+ } catch (HadoopIllegalArgumentException e) {
shutdown();
- LOG.fatal("Failed to start secondary namenode. ", e);
throw e;
}
}
@@ -216,8 +212,9 @@ public class SecondaryNameNode implement
nameNodeAddr = NameNode.getServiceAddress(conf, true);
this.conf = conf;
- this.namenode = new NamenodeProtocolTranslatorPB(nameNodeAddr, conf,
- UserGroupInformation.getCurrentUser());
+ this.namenode = NameNodeProxies.createNonHAProxy(conf, nameNodeAddr,
+ NamenodeProtocol.class, UserGroupInformation.getCurrentUser(),
+ true).getProxy();
// initialize checkpoint directories
fsName = getInfoServer();
@@ -231,16 +228,8 @@ public class SecondaryNameNode implement
namesystem = new FSNamesystem(conf, checkpointImage);
// Initialize other scheduling parameters from the configuration
- checkpointCheckPeriod = conf.getLong(
- DFS_NAMENODE_CHECKPOINT_CHECK_PERIOD_KEY,
- DFS_NAMENODE_CHECKPOINT_CHECK_PERIOD_DEFAULT);
-
- checkpointPeriod = conf.getLong(DFS_NAMENODE_CHECKPOINT_PERIOD_KEY,
- DFS_NAMENODE_CHECKPOINT_PERIOD_DEFAULT);
- checkpointTxnCount = conf.getLong(DFS_NAMENODE_CHECKPOINT_TXNS_KEY,
- DFS_NAMENODE_CHECKPOINT_TXNS_DEFAULT);
- warnForDeprecatedConfigs(conf);
-
+ checkpointConf = new CheckpointConf(conf);
+
// initialize the webserver for uploading files.
// Kerberized SSL servers must be run from the host principal...
UserGroupInformation httpUGI =
@@ -296,21 +285,9 @@ public class SecondaryNameNode implement
conf.set(DFS_NAMENODE_SECONDARY_HTTP_ADDRESS_KEY, infoBindAddress + ":" +infoPort);
LOG.info("Secondary Web-server up at: " + infoBindAddress + ":" +infoPort);
LOG.info("Secondary image servlet up at: " + infoBindAddress + ":" + imagePort);
- LOG.info("Checkpoint Period :" + checkpointPeriod + " secs " +
- "(" + checkpointPeriod/60 + " min)");
- LOG.info("Log Size Trigger :" + checkpointTxnCount + " txns");
- }
-
- static void warnForDeprecatedConfigs(Configuration conf) {
- for (String key : ImmutableList.of(
- "fs.checkpoint.size",
- "dfs.namenode.checkpoint.size")) {
- if (conf.get(key) != null) {
- LOG.warn("Configuration key " + key + " is deprecated! Ignoring..." +
- " Instead please specify a value for " +
- DFS_NAMENODE_CHECKPOINT_TXNS_KEY);
- }
- }
+ LOG.info("Checkpoint Period :" + checkpointConf.getPeriod() + " secs " +
+ "(" + checkpointConf.getPeriod()/60 + " min)");
+ LOG.info("Log Size Trigger :" + checkpointConf.getTxnCount() + " txns");
}
/**
@@ -332,36 +309,24 @@ public class SecondaryNameNode implement
}
public void run() {
- if (UserGroupInformation.isSecurityEnabled()) {
- UserGroupInformation ugi = null;
- try {
- ugi = UserGroupInformation.getLoginUser();
- } catch (IOException e) {
- LOG.error("Exception while getting login user", e);
- e.printStackTrace();
- Runtime.getRuntime().exit(-1);
- }
- ugi.doAs(new PrivilegedAction<Object>() {
+ SecurityUtil.doAsLoginUserOrFatal(
+ new PrivilegedAction<Object>() {
@Override
public Object run() {
doWork();
return null;
}
});
- } else {
- doWork();
- }
}
//
// The main work loop
//
public void doWork() {
-
//
// Poll the Namenode (once every checkpointCheckPeriod seconds) to find the
// number of transactions in the edit log that haven't yet been checkpointed.
//
- long period = Math.min(checkpointCheckPeriod, checkpointPeriod);
+ long period = checkpointConf.getCheckPeriod();
while (shouldRun) {
try {
@@ -380,7 +345,7 @@ public class SecondaryNameNode implement
long now = System.currentTimeMillis();
if (shouldCheckpointBasedOnCount() ||
- now >= lastCheckpointTime + 1000 * checkpointPeriod) {
+ now >= lastCheckpointTime + 1000 * checkpointConf.getPeriod()) {
doCheckpoint();
lastCheckpointTime = now;
}
@@ -471,19 +436,10 @@ public class SecondaryNameNode implement
}
String configuredAddress = DFSUtil.getInfoServer(null, conf, true);
- InetSocketAddress sockAddr = NetUtils.createSocketAddr(configuredAddress);
- if (sockAddr.getAddress().isAnyLocalAddress()) {
- if(UserGroupInformation.isSecurityEnabled()) {
- throw new IOException("Cannot use a wildcard address with security. " +
- "Must explicitly set bind address for Kerberos");
- }
- return fsName.getHost() + ":" + sockAddr.getPort();
- } else {
- if(LOG.isDebugEnabled()) {
- LOG.debug("configuredAddress = " + configuredAddress);
- }
- return configuredAddress;
- }
+ String address = DFSUtil.substituteForWildcardAddress(configuredAddress,
+ fsName.getHost());
+ LOG.debug("Will connect to NameNode at HTTP address: " + address);
+ return address;
}
/**
@@ -574,13 +530,13 @@ public class SecondaryNameNode implement
switch (opts.getCommand()) {
case CHECKPOINT:
long count = countUncheckpointedTxns();
- if (count > checkpointTxnCount ||
+ if (count > checkpointConf.getTxnCount() ||
opts.shouldForceCheckpoint()) {
doCheckpoint();
} else {
System.err.println("EditLog size " + count + " transactions is " +
"smaller than configured checkpoint " +
- "interval " + checkpointTxnCount + " transactions.");
+ "interval " + checkpointConf.getTxnCount() + " transactions.");
System.err.println("Skipping checkpoint.");
}
break;
@@ -626,7 +582,7 @@ public class SecondaryNameNode implement
}
boolean shouldCheckpointBasedOnCount() throws IOException {
- return countUncheckpointedTxns() >= checkpointTxnCount;
+ return countUncheckpointedTxns() >= checkpointConf.getTxnCount();
}
/**
@@ -642,7 +598,13 @@ public class SecondaryNameNode implement
StringUtils.startupShutdownMessage(SecondaryNameNode.class, argv, LOG);
Configuration tconf = new HdfsConfiguration();
- SecondaryNameNode secondary = new SecondaryNameNode(tconf, opts);
+ SecondaryNameNode secondary = null;
+ try {
+ secondary = new SecondaryNameNode(tconf, opts);
+ } catch (IOException ioe) {
+ LOG.fatal("Failed to start secondary namenode", ioe);
+ System.exit(-1);
+ }
if (opts.getCommand() != null) {
int ret = secondary.processStartupCommand(opts);
@@ -759,7 +721,7 @@ public class SecondaryNameNode implement
*/
CheckpointStorage(Configuration conf,
Collection<URI> imageDirs,
- Collection<URI> editsDirs) throws IOException {
+ List<URI> editsDirs) throws IOException {
super(conf, imageDirs, editsDirs);
// the 2NN never writes edits -- it only downloads them. So
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/TransferFsImage.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/TransferFsImage.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/TransferFsImage.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/TransferFsImage.java Sat Mar 3 00:42:49 2012
@@ -24,8 +24,11 @@ import java.security.MessageDigest;
import java.util.List;
import java.lang.Math;
+import javax.servlet.http.HttpServletResponse;
+
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.security.SecurityUtil;
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.hdfs.server.namenode.NNStorage.NameNodeDirType;
@@ -41,7 +44,8 @@ import com.google.common.collect.Lists;
/**
* This class provides fetching a specified file from the NameNode.
*/
-class TransferFsImage {
+@InterfaceAudience.Private
+public class TransferFsImage {
public final static String CONTENT_LENGTH = "Content-Length";
public final static String MD5_HEADER = "X-MD5-Digest";
@@ -103,7 +107,7 @@ class TransferFsImage {
* @param storage the storage directory to transfer the image from
* @param txid the transaction ID of the image to be uploaded
*/
- static void uploadImageFromStorage(String fsName,
+ public static void uploadImageFromStorage(String fsName,
InetSocketAddress imageListenAddress,
NNStorage storage, long txid) throws IOException {
@@ -111,7 +115,20 @@ class TransferFsImage {
txid, imageListenAddress, storage);
// this doesn't directly upload an image, but rather asks the NN
// to connect back to the 2NN to download the specified image.
- TransferFsImage.getFileClient(fsName, fileid, null, null, false);
+ try {
+ TransferFsImage.getFileClient(fsName, fileid, null, null, false);
+ } catch (HttpGetFailedException e) {
+ if (e.getResponseCode() == HttpServletResponse.SC_CONFLICT) {
+ // this is OK - this means that a previous attempt to upload
+ // this checkpoint succeeded even though we thought it failed.
+ LOG.info("Image upload with txid " + txid +
+ " conflicted with a previous image upload to the " +
+ "same NameNode. Continuing...", e);
+ return;
+ } else {
+ throw e;
+ }
+ }
LOG.info("Uploaded image with txid " + txid + " to namenode at " +
fsName);
}
@@ -194,10 +211,11 @@ class TransferFsImage {
HttpURLConnection connection = (HttpURLConnection) url.openConnection();
if (connection.getResponseCode() != HttpURLConnection.HTTP_OK) {
- throw new IOException(
+ throw new HttpGetFailedException(
"Image transfer servlet at " + url +
" failed with status code " + connection.getResponseCode() +
- "\nResponse message:\n" + connection.getResponseMessage());
+ "\nResponse message:\n" + connection.getResponseMessage(),
+ connection);
}
long advertisedSize;
@@ -289,5 +307,19 @@ class TransferFsImage {
String header = connection.getHeaderField(MD5_HEADER);
return (header != null) ? new MD5Hash(header) : null;
}
+
+ public static class HttpGetFailedException extends IOException {
+ private static final long serialVersionUID = 1L;
+ private final int responseCode;
+
+ HttpGetFailedException(String msg, HttpURLConnection connection) throws IOException {
+ super(msg);
+ this.responseCode = connection.getResponseCode();
+ }
+
+ public int getResponseCode() {
+ return responseCode;
+ }
+ }
}
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/UnsupportedActionException.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/UnsupportedActionException.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/UnsupportedActionException.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/UnsupportedActionException.java Sat Mar 3 00:42:49 2012
@@ -32,8 +32,7 @@ public class UnsupportedActionException
/** for java.io.Serializable */
private static final long serialVersionUID = 1L;
- public UnsupportedActionException(String action) {
- super("Action " + action + "() is not supported.");
+ public UnsupportedActionException(String msg) {
+ super(msg);
}
-
}
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/DatanodeProtocol.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/DatanodeProtocol.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/DatanodeProtocol.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/DatanodeProtocol.java Sat Mar 3 00:42:49 2012
@@ -22,8 +22,8 @@ import java.io.*;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.hdfs.DFSConfigKeys;
-import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
import org.apache.hadoop.hdfs.protocol.DatanodeID;
+import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
import org.apache.hadoop.security.KerberosInfo;
@@ -93,7 +93,7 @@ public interface DatanodeProtocol {
* sendHeartbeat() tells the NameNode that the DataNode is still
* alive and well. Includes some status info, too.
* It also gives the NameNode a chance to return
- * an array of "DatanodeCommand" objects.
+ * an array of "DatanodeCommand" objects in HeartbeatResponse.
* A DatanodeCommand tells the DataNode to invalidate local block(s),
* or to copy them to other DataNodes, etc.
* @param registration datanode registration information
@@ -103,7 +103,7 @@ public interface DatanodeProtocol {
* @param failedVolumes number of failed volumes
* @throws IOException on error
*/
- public DatanodeCommand[] sendHeartbeat(DatanodeRegistration registration,
+ public HeartbeatResponse sendHeartbeat(DatanodeRegistration registration,
StorageReport[] reports,
int xmitsInProgress,
int xceiverCount,
@@ -118,7 +118,8 @@ public interface DatanodeProtocol {
* @param registration
* @param poolId - the block pool ID for the blocks
* @param reports - report of blocks per storage
- * Each block is represented as 2 longs.
+ * Each finalized block is represented as 3 longs. Each under-
+ * construction replica is represented as 4 longs.
* This is done instead of Block[] to reduce memory used by block reports.
*
* @return - the next command for DN to process.
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/NamenodeProtocol.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/NamenodeProtocol.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/NamenodeProtocol.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/NamenodeProtocol.java Sat Mar 3 00:42:49 2012
@@ -101,10 +101,7 @@ public interface NamenodeProtocol {
* call fails if the file system is in SafeMode.
* @throws IOException
* @return a unique token to identify this transaction.
- * @deprecated
- * See {@link org.apache.hadoop.hdfs.server.namenode.SecondaryNameNode}
*/
- @Deprecated
public CheckpointSignature rollEditLog() throws IOException;
/**
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/NamenodeProtocols.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/NamenodeProtocols.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/NamenodeProtocols.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/NamenodeProtocols.java Sat Mar 3 00:42:49 2012
@@ -19,6 +19,7 @@
package org.apache.hadoop.hdfs.server.protocol;
import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.ha.HAServiceProtocol;
import org.apache.hadoop.hdfs.protocol.ClientProtocol;
import org.apache.hadoop.security.authorize.RefreshAuthorizationPolicyProtocol;
import org.apache.hadoop.security.RefreshUserMappingsProtocol;
@@ -32,5 +33,6 @@ public interface NamenodeProtocols
NamenodeProtocol,
RefreshAuthorizationPolicyProtocol,
RefreshUserMappingsProtocol,
- GetUserMappingsProtocol {
+ GetUserMappingsProtocol,
+ HAServiceProtocol {
}
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/ReceivedDeletedBlockInfo.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/ReceivedDeletedBlockInfo.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/ReceivedDeletedBlockInfo.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/ReceivedDeletedBlockInfo.java Sat Mar 3 00:42:49 2012
@@ -25,22 +25,47 @@ import java.io.IOException;
import org.apache.hadoop.hdfs.protocol.Block;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableUtils;
/**
- * A data structure to store Block and delHints together, used to send
- * received/deleted ACKs.
+ * A data structure to store the blocks in an incremental block report.
*/
public class ReceivedDeletedBlockInfo implements Writable {
Block block;
+ BlockStatus status;
String delHints;
- public final static String TODELETE_HINT = "-";
+ public static enum BlockStatus {
+ RECEIVING_BLOCK(1),
+ RECEIVED_BLOCK(2),
+ DELETED_BLOCK(3);
+
+ private final int code;
+ BlockStatus(int code) {
+ this.code = code;
+ }
+
+ public int getCode() {
+ return code;
+ }
+
+ public static BlockStatus fromCode(int code) {
+ for (BlockStatus bs : BlockStatus.values()) {
+ if (bs.code == code) {
+ return bs;
+ }
+ }
+ return null;
+ }
+ }
public ReceivedDeletedBlockInfo() {
}
- public ReceivedDeletedBlockInfo(Block blk, String delHints) {
+ public ReceivedDeletedBlockInfo(
+ Block blk, BlockStatus status, String delHints) {
this.block = blk;
+ this.status = status;
this.delHints = delHints;
}
@@ -60,13 +85,19 @@ public class ReceivedDeletedBlockInfo im
this.delHints = hints;
}
+ public BlockStatus getStatus() {
+ return status;
+ }
+
public boolean equals(Object o) {
if (!(o instanceof ReceivedDeletedBlockInfo)) {
return false;
}
ReceivedDeletedBlockInfo other = (ReceivedDeletedBlockInfo) o;
return this.block.equals(other.getBlock())
- && this.delHints.equals(other.delHints);
+ && this.status == other.status
+ && (this.delHints == other.delHints ||
+ this.delHints != null && this.delHints.equals(other.delHints));
}
public int hashCode() {
@@ -79,23 +110,30 @@ public class ReceivedDeletedBlockInfo im
}
public boolean isDeletedBlock() {
- return delHints.equals(TODELETE_HINT);
+ return status == BlockStatus.DELETED_BLOCK;
}
@Override
public void write(DataOutput out) throws IOException {
this.block.write(out);
- Text.writeString(out, this.delHints);
+ WritableUtils.writeVInt(out, this.status.code);
+ if (this.status == BlockStatus.DELETED_BLOCK) {
+ Text.writeString(out, this.delHints);
+ }
}
@Override
public void readFields(DataInput in) throws IOException {
this.block = new Block();
this.block.readFields(in);
- this.delHints = Text.readString(in);
+ this.status = BlockStatus.fromCode(WritableUtils.readVInt(in));
+ if (this.status == BlockStatus.DELETED_BLOCK) {
+ this.delHints = Text.readString(in);
+ }
}
public String toString() {
- return block.toString() + ", delHint: " + delHints;
+ return block.toString() + ", status: " + status +
+ ", delHint: " + delHints;
}
}
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSAdmin.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSAdmin.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSAdmin.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSAdmin.java Sat Mar 3 00:42:49 2012
@@ -38,19 +38,20 @@ import org.apache.hadoop.hdfs.DFSConfigK
import org.apache.hadoop.hdfs.DFSUtil;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.HdfsConfiguration;
+import org.apache.hadoop.hdfs.NameNodeProxies;
import org.apache.hadoop.hdfs.protocol.ClientDatanodeProtocol;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.hdfs.protocol.HdfsConstants.DatanodeReportType;
import org.apache.hadoop.hdfs.protocol.HdfsConstants.UpgradeAction;
-import org.apache.hadoop.hdfs.protocolPB.RefreshAuthorizationPolicyProtocolClientSideTranslatorPB;
-import org.apache.hadoop.hdfs.protocolPB.RefreshUserMappingsProtocolClientSideTranslatorPB;
import org.apache.hadoop.hdfs.server.common.UpgradeStatusReport;
import org.apache.hadoop.hdfs.server.namenode.NameNode;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.RemoteException;
import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.RefreshUserMappingsProtocol;
import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.authorize.RefreshAuthorizationPolicyProtocol;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.ToolRunner;
@@ -791,9 +792,9 @@ public class DFSAdmin extends FsShell {
conf.get(DFSConfigKeys.DFS_NAMENODE_USER_NAME_KEY, ""));
// Create the client
- RefreshAuthorizationPolicyProtocolClientSideTranslatorPB refreshProtocol =
- new RefreshAuthorizationPolicyProtocolClientSideTranslatorPB(
- NameNode.getAddress(conf), getUGI(), conf);
+ RefreshAuthorizationPolicyProtocol refreshProtocol =
+ NameNodeProxies.createProxy(conf, FileSystem.getDefaultUri(conf),
+ RefreshAuthorizationPolicyProtocol.class).getProxy();
// Refresh the authorization policy in-effect
refreshProtocol.refreshServiceAcl();
@@ -817,9 +818,9 @@ public class DFSAdmin extends FsShell {
conf.get(DFSConfigKeys.DFS_NAMENODE_USER_NAME_KEY, ""));
// Create the client
- RefreshUserMappingsProtocolClientSideTranslatorPB refreshProtocol =
- new RefreshUserMappingsProtocolClientSideTranslatorPB(
- NameNode.getAddress(conf), getUGI(), conf);
+ RefreshUserMappingsProtocol refreshProtocol =
+ NameNodeProxies.createProxy(conf, FileSystem.getDefaultUri(conf),
+ RefreshUserMappingsProtocol.class).getProxy();
// Refresh the user-to-groups mappings
refreshProtocol.refreshUserToGroupsMappings();
@@ -844,9 +845,9 @@ public class DFSAdmin extends FsShell {
conf.get(DFSConfigKeys.DFS_NAMENODE_USER_NAME_KEY, ""));
// Create the client
- RefreshUserMappingsProtocolClientSideTranslatorPB refreshProtocol =
- new RefreshUserMappingsProtocolClientSideTranslatorPB(
- NameNode.getAddress(conf), getUGI(), conf);
+ RefreshUserMappingsProtocol refreshProtocol =
+ NameNodeProxies.createProxy(conf, FileSystem.getDefaultUri(conf),
+ RefreshUserMappingsProtocol.class).getProxy();
// Refresh the user-to-groups mappings
refreshProtocol.refreshSuperUserGroupsConfiguration();
Modified: hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSck.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSck.java?rev=1296534&r1=1296533&r2=1296534&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSck.java (original)
+++ hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSck.java Sat Mar 3 00:42:49 2012
@@ -32,11 +32,13 @@ import org.apache.hadoop.classification.
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.hdfs.server.namenode.NameNode;
-import org.apache.hadoop.hdfs.server.namenode.NamenodeFsck;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.DFSClient;
import org.apache.hadoop.hdfs.DFSUtil;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.HdfsConfiguration;
+import org.apache.hadoop.hdfs.server.namenode.NamenodeFsck;
+import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.security.Krb5AndCertsSslSocketConnector;
import org.apache.hadoop.security.SecurityUtil;
import org.apache.hadoop.security.UserGroupInformation;
@@ -204,8 +206,9 @@ public class DFSck extends Configured im
* Derive the namenode http address from the current file system,
* either default or as set by "-fs" in the generic options.
* @return Returns http address or null if failure.
+ * @throws IOException if we can't determine the active NN address
*/
- private String getCurrentNamenodeAddress() {
+ private String getCurrentNamenodeAddress() throws IOException {
//String nnAddress = null;
Configuration conf = getConf();
@@ -222,16 +225,21 @@ public class DFSck extends Configured im
System.err.println("FileSystem is " + fs.getUri());
return null;
}
+
+ // force client address resolution.
+ fs.exists(new Path("/"));
+
+ // Derive the nameservice ID from the filesystem connection. The URI may
+ // have been provided by a human, the server name may be aliased, or there
+ // may be multiple possible actual addresses (e.g. in an HA setup) so
+ // compare InetSocketAddresses instead of URI strings, and test against both
+ // possible configurations of RPC address (DFS_NAMENODE_RPC_ADDRESS_KEY and
+ // DFS_NAMENODE_SERVICE_RPC_ADDRESS_KEY).
DistributedFileSystem dfs = (DistributedFileSystem) fs;
-
- // Derive the nameservice ID from the filesystem URI.
- // The URI may have been provided by a human, and the server name may be
- // aliased, so compare InetSocketAddresses instead of URI strings, and
- // test against both possible variants of RPC address.
- InetSocketAddress namenode =
- NameNode.getAddress(dfs.getUri().getAuthority());
+ DFSClient dfsClient = dfs.getClient();
+ InetSocketAddress addr = RPC.getServerAddress(dfsClient.getNamenode());
- return DFSUtil.getInfoServer(namenode, conf, true);
+ return DFSUtil.getInfoServer(addr, conf, true);
}
private int doWork(final String[] args) throws IOException {